Runtime API Reference

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

Authorization: Bearer <DEEPLINE_API_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/"
    }
  }'

{
  "workflowId": "play_run_...",
  "apiVersion": 2,
  "status": "running",
  "dashboardUrl": "https://code.deepline.com/dashboard/plays/..."
}

curl "$DEEPLINE_HOST_URL/api/v2/runs/$WORKFLOW_ID?full=true" \
  -H "Authorization: Bearer $DEEPLINE_API_KEY"

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

curl -X POST "$DEEPLINE_HOST_URL/api/v2/runs/$WORKFLOW_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


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",
}

start = requests.post(
    f"{BASE_URL}/api/v2/plays/run",
    headers=headers,
    json={
        "name": "prebuilt/person-linkedin-to-email",
        "input": {
            "linkedin_url": "https://www.linkedin.com/in/example-person/",
        },
    },
    timeout=30,
)
start.raise_for_status()
workflow_id = start.json()["workflowId"]

while True:
    status = requests.get(
        f"{BASE_URL}/api/v2/runs/{workflow_id}",
        headers=headers,
        timeout=30,
    )
    status.raise_for_status()
    body = status.json()
    if body.get("status") in {"completed", "failed", "cancelled"}:
        with open("person-email-result.json", "w") as f:
            json.dump(body, f, indent=2)
        print(body)
        break
    time.sleep(2)