Runtime API Reference

export DEEPLINE_HOST_URL="${DEEPLINE_HOST_URL:-https://code.deepline.com}"
export DEEPLINE_API_KEY="dl_workspace_key"

curl -X POST "$DEEPLINE_HOST_URL/api/v2/plays/run" \
  -H "Authorization: Bearer $DEEPLINE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "prebuilt/person-linkedin-to-email",
    "input": {
      "linkedin_url": "https://www.linkedin.com/in/example-person/"
    }
  }'

{
  "run": {
    "id": "play/prebuilt/person-linkedin-to-email/run/example",
    "playName": "person-linkedin-to-email",
    "status": "running"
  },
  "outputs": {}
}

ENCODED_RUN_ID=$(jq -er '.run.id | @uri' launch.json)
curl "$DEEPLINE_HOST_URL/api/v2/runs/$ENCODED_RUN_ID" \
  -H "Authorization: Bearer $DEEPLINE_API_KEY"

curl -N "$DEEPLINE_HOST_URL/api/v2/runs/$ENCODED_RUN_ID/tail?mode=cli" \
  -H "Authorization: Bearer $DEEPLINE_API_KEY" \
  -H "Accept: text/event-stream"

curl -X POST "$DEEPLINE_HOST_URL/api/v2/runs/$ENCODED_RUN_ID/stop" \
  -H "Authorization: Bearer $DEEPLINE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"reason":"caller cancelled"}'

import os
import time
import json
import requests
from pathlib import Path
from urllib.parse import quote


def load_deepline_env(path=".env.deepline"):
    values = {}
    if not os.path.exists(path):
        return values
    with open(path) as env_file:
        for line in env_file:
            stripped = line.strip()
            if not stripped or stripped.startswith("#") or "=" not in stripped:
                continue
            key, value = stripped.split("=", 1)
            values[key.strip()] = value.strip().strip('"').strip("'")
    return values


deepline_env = load_deepline_env()
BASE_URL = os.environ.get(
    "DEEPLINE_HOST_URL",
    deepline_env.get("DEEPLINE_HOST_URL", "https://code.deepline.com"),
)
API_KEY = os.environ.get("DEEPLINE_API_KEY", deepline_env.get("DEEPLINE_API_KEY"))
if not API_KEY:
    raise RuntimeError("Missing DEEPLINE_API_KEY in .env.deepline")

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

# Reuse this file only with the same workspace authorization and input.
run_file = Path("person-email-run.json")
if run_file.exists():
    saved = json.loads(run_file.read_text())
    if saved["base_url"] != BASE_URL:
        raise RuntimeError("Saved run belongs to another Deepline origin")
    run_id = saved["run_id"]
    if run_id is None:
        raise RuntimeError("Prior submission outcome is unknown; investigate before starting new work")
else:
    # A lost launch response must not cause automatic resubmission.
    run_file.write_text(json.dumps({"base_url": BASE_URL, "run_id": None}))
    start = requests.post(
        f"{BASE_URL}/api/v2/plays/run",
        headers=headers,
        json={
            "name": "prebuilt/person-linkedin-to-email",
            "input": {
                # Replace this illustrative URL before starting provider work.
                "linkedin_url": "https://www.linkedin.com/in/example-person/",
            },
        },
        timeout=30,
    )
    start.raise_for_status()
    run_id = start.json()["run"]["id"]
    if not isinstance(run_id, str) or not run_id:
        raise RuntimeError("Launch response has no nonempty run ID")
    run_file.write_text(json.dumps({"base_url": BASE_URL, "run_id": run_id}))

if not isinstance(run_id, str) or not run_id:
    raise RuntimeError("The saved response must contain a nonempty run ID")

deadline = time.monotonic() + 600
while True:
    response = requests.get(
        f"{BASE_URL}/api/v2/runs/{quote(run_id, safe='')}",
        headers=headers,
        timeout=30,
    )
    response.raise_for_status()
    body = response.json()
    run_status = body["run"]["status"]
    if run_status in {"completed", "failed", "cancelled"}:
        Path("person-email-result.json").write_text(json.dumps(body, indent=2))
        print(f"Run {run_id}: {run_status}; saved person-email-result.json")
        if run_status != "completed":
            raise RuntimeError(f"Run ended with {run_status}; inspect the saved result")
        break
    if time.monotonic() >= deadline:
        raise TimeoutError(
            f"Local wait ended for {run_id}; remote work was not cancelled. "
            "Resume with the same saved run file and workspace authorization."
        )
    time.sleep(min(2, max(0, deadline - time.monotonic())))