Pantheon COO OS — API & integration

Quick reference for developers. OpenAPI: /docs

1. Quick start (curl)

curl -X POST http://localhost:8002/execute \
  -H "Content-Type: application/json" \
  -d '{"command": "List files in workspace"}'

curl http://localhost:8002/tasks/{task_id}

curl -N http://localhost:8002/tasks/{task_id}/stream

curl http://localhost:8002/health

2. Authentication

JWT: Authorization: Bearer <token> after POST /auth/login.

API key (legacy): X-COO-API-Key when AUTH_MODE=apikey.

curl -X POST http://localhost:8002/auth/login \
  -H "Content-Type: application/json" \
  -d '{"email":"you@example.com","password":"********"}'

3. Execute a task (poll in JavaScript)

async function run(cmd) {
  const r = await fetch('/execute', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'Authorization': 'Bearer ' + token },
    body: JSON.stringify({ command: cmd })
  });
  const { task_id } = await r.json();
  while (true) {
    const t = await (await fetch('/tasks/' + task_id, { headers: { Authorization: 'Bearer ' + token } })).json();
    if (t.status === 'done' || t.status === 'failed') return t;
    await new Promise(x => setTimeout(x, 2000));
  }
}

4. SSE streaming (EventSource)

const es = new EventSource('/tasks/' + taskId + '/stream');
es.onmessage = (ev) => console.log(JSON.parse(ev.data));

5. Webhooks (outbound)

Subscribe with POST /webhooks (authenticated). Verify X-Pantheon-Signature: sha256=<hmac> on the raw body using your secret.

6. Rate limits (by plan)

See GET /usage for your current window.

7. Error codes

• 401 — missing/invalid auth
• 403 — forbidden (e.g. admin routes)
• 429 — rate limited
• 503 — dependency not configured (billing, etc.)

8. Python SDK example

import requests

class PantheonCOO:
    def __init__(self, api_key, base_url="https://your-deployment.railway.app"):
        self.headers = {"X-COO-API-Key": api_key}
        self.base_url = base_url.rstrip("/")

    def execute(self, command: str) -> dict:
        r = requests.post(
            f"{self.base_url}/execute",
            json={"command": command},
            headers=self.headers,
            timeout=60,
        )
        r.raise_for_status()
        return r.json()

    def get_task(self, task_id: str) -> dict:
        r = requests.get(
            f"{self.base_url}/tasks/{task_id}",
            headers=self.headers,
            timeout=30,
        )
        r.raise_for_status()
        return r.json()

    def wait_for_completion(self, task_id, timeout=120):
        import time
        start = time.time()
        while time.time() - start < timeout:
            task = self.get_task(task_id)
            if task["status"] in ["done", "failed"]:
                return task
            time.sleep(2)
        return None