SDK Python

Python integration guide with request/response parity snippets for critical DCP endpoints.

Install

pip install dc1

Renter flow (me -> submit -> logs -> output)

import json
import requests

BASE_URL = "https://api.dcp.sa"
RENTER_KEY = "dc1-renter-your-key"

headers = {
    "Content-Type": "application/json",
    "x-renter-key": RENTER_KEY,
}

# ) Validate auth
me = requests.get(f"{BASE_URL}/api/renters/me", params={"key": RENTER_KEY}, timeout=)
me.raise_for_status()
print("renter:", me.json().get("renter", {}).get("email"))

# ) Submit job
submit_payload = {
    "provider_id": ,
    "job_type": "llm_inference",
    "duration_minutes": ,
    "max_duration_seconds": ,
    "container_spec": {"image_type": "vllm-serve"},
    "params": {
        "model": "TinyLlama/TinyLlama-.1B-Chat-v1.",
        "prompt": "Give  bullets on DCP.",
    },
}

submit = requests.post(f"{BASE_URL}/api/jobs/submit", headers=headers, json=submit_payload, timeout=)
submit.raise_for_status()
job_id = submit.json()["job"]["job_id"]
print("job_id:", job_id)

# ) Stream logs (SSE)
stream_url = f"{BASE_URL}/api/jobs/{job_id}/logs/stream"
with requests.get(stream_url, params={"key": RENTER_KEY}, stream=True, timeout=) as resp:
    resp.raise_for_status()
    for raw in resp.iter_lines(decode_unicode=True):
        if not raw or not raw.startswith("data: "):
            continue
        evt = json.loads(raw[:])
        if evt.get("type") == "log":
            print("[log]", evt.get("line"))
        if evt.get("type") == "end":
            print("[end]", evt.get("status"))
            break

# ) Fetch output
output = requests.get(f"{BASE_URL}/api/jobs/{job_id}/output", params={"key": RENTER_KEY}, timeout=)
print(output.status_code, output.json())

Parity snippets (raw requests)

`POST /api/renters/register`

register = requests.post(
    "https://api.dcp.sa/api/renters/register",
    json={
        "name": "Fatima Al-Saud",
        "email": "fatima@example.sa",
        "organization": "Riyadh AI Lab",
        "use_case": "llm_inference",
        "phone": "+",
    },
    timeout=,
)
register.raise_for_status()
print(register.json())

`POST /api/providers/heartbeat`

heartbeat = requests.post(
    "https://api.dcp.sa/api/providers/heartbeat",
    json={
        "api_key": "dc1-provider-abc123...",
        "gpu_status": {
            "gpu_name": "NVIDIA RTX ",
            "gpu_vram_mib": ,
            "gpu_util_pct": ,
            "temp_c": ,
            "daemon_version": ".3.",
        },
    },
    timeout=,
)
heartbeat.raise_for_status()
print(heartbeat.json())

`GET /api/admin/dashboard`

admin = requests.get(
    "https://api.dcp.sa/api/admin/dashboard",
    headers={"x-admin-token": "<DC1_ADMIN_TOKEN>"},
    timeout=,
)
admin.raise_for_status()
print(admin.json().get("stats"))

Notes

  • Error payloads are always JSON: `{ "error": "..." }`.
  • Job submit auth: `x-renter-key` header (or `?key=` / `?renter_key=` query fallback).
  • Admin auth: `x-admin-token` header (Bearer token also accepted).