UcoWorker Cloud Agent API

create ──> chat ──> [running] ──> done
                    │                       │
                    │  (stream drops?)      │
                    │      ↓                │
                    │  poll /status ────────│
                    │                       │
                    └── send another /chat ─┘  (resume)

                         delete ───────────>   (archive)

import httpx, json, os

BASE = "https://api.ucoworker.com/v1/agent/cloud"
AUTH = {"Authorization": f"Bearer {os.environ['UCOWORKER_API_KEY']}"}

# Create a conversation
conv = httpx.post(f"{BASE}/conversations", headers=AUTH,
                  json={"agent": "general"}).json()
conv_id = conv["id"]

# Stream the response
answer = ""
with httpx.stream("POST", f"{BASE}/conversations/{conv_id}/chat",
                   headers=AUTH,
                   files=[("message", (None, "What is 2+2?"))],
                   timeout=httpx.Timeout(None, connect=15)) as resp:
    for line in resp.iter_lines():
        if not line.strip():
            continue
        kind, _, payload = line.partition(":")
        if kind == "0":
            chunk = json.loads(payload)
            answer += chunk
            print(chunk, end="", flush=True)
        elif kind in ("d", "3"):
            break

# Confirm via status
status = httpx.get(f"{BASE}/conversations/{conv_id}/status",
                   headers=AUTH).json()
final = status.get("answer") or answer

import httpx, json, time, os

BASE = "https://api.ucoworker.com/v1/agent/cloud"
AUTH = {"Authorization": f"Bearer {os.environ['UCOWORKER_API_KEY']}"}
TERMINAL = {"done", "error", "timeout", "cancelled"}

def chat_and_wait(conv_id: str, message: str, model: str | None = None) -> str:
    """Send a message, stream the response, poll if the stream drops."""
    parts = [("message", (None, message))]
    if model:
        parts.append(("model", (None, model)))

    answer = ""
    finished = False

    try:
        with httpx.stream("POST", f"{BASE}/conversations/{conv_id}/chat",
                          headers=AUTH, files=parts,
                          timeout=httpx.Timeout(None, connect=15)) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if not line.strip():
                    continue
                kind, _, payload = line.partition(":")
                if kind == "0":
                    answer += json.loads(payload)
                elif kind in ("d", "3"):
                    finished = True
                    break
    except (httpx.HTTPError, httpx.StreamError):
        pass  # stream dropped -- fall through to poll

    if not finished:
        for _ in range(400):
            status = httpx.get(f"{BASE}/conversations/{conv_id}/status",
                               headers=AUTH).json()
            if status.get("status") in TERMINAL:
                return status.get("answer") or answer
            time.sleep(1.5)

    return answer

import json, os, httpx

STATE_FILE = ".agent_state.json"

def get_or_create_conv(label: str, agent: str = "general") -> str:
    """Reuse an existing conversation or create a new one."""
    state = {}
    if os.path.exists(STATE_FILE):
        state = json.load(open(STATE_FILE))

    conv_id = state.get(label)
    if conv_id:
        return conv_id

    resp = httpx.post(f"{BASE}/conversations", headers=AUTH,
                      json={"agent": agent, "title": label})
    resp.raise_for_status()
    conv_id = resp.json()["id"]

    state[label] = conv_id
    json.dump(state, open(STATE_FILE, "w"), indent=2)
    return conv_id

# First run: creates a new conversation
conv_id = get_or_create_conv("daily-research")
answer = chat_and_wait(conv_id, "What happened in tech today?")

# Next run (hours/days later): resumes the same conversation
conv_id = get_or_create_conv("daily-research")
answer = chat_and_wait(conv_id, "Any updates since last time?")

# Tell the agent to write output
chat_and_wait(conv_id, 'Write a JSON summary to output/summary.json')

import time; time.sleep(2)  # brief wait for file persistence

# Try the expected path
resp = httpx.get(
    f"{BASE}/conversations/{conv_id}/workspace/files/output/summary.json",
    headers=AUTH
)

if resp.status_code == 404:
    # Fallback: list workspace and find it
    tree = httpx.get(
        f"{BASE}/conversations/{conv_id}/workspace/files",
        headers=AUTH
    ).json()
    for entry in tree.get("entries", []):
        if entry["path"].endswith("summary.json"):
            resp = httpx.get(
                f"{BASE}/conversations/{conv_id}/workspace/files/{entry['path']}",
                headers=AUTH
            )
            break

if resp.status_code == 200:
    data = resp.json()
    print(json.dumps(data, indent=2))

import asyncio, httpx, json, os

BASE = "https://api.ucoworker.com/v1/agent/cloud"
MAX_CONCURRENT = 5

async def run_task(client, sem, name, prompt):
    async with sem:
        resp = await client.post(f"{BASE}/conversations",
                                 json={"agent": "general", "title": name})
        conv_id = resp.json()["id"]

        answer = ""
        async with client.stream(
            "POST", f"{BASE}/conversations/{conv_id}/chat",
            files=[("message", (None, prompt))],
            timeout=httpx.Timeout(None, connect=15)
        ) as r:
            async for line in r.aiter_lines():
                kind, _, payload = line.partition(":")
                if kind == "0":
                    answer += json.loads(payload)
                elif kind in ("d", "3"):
                    break
        return {"task": name, "answer": answer}

async def main():
    auth = {"Authorization": f"Bearer {os.environ['UCOWORKER_API_KEY']}"}
    sem = asyncio.Semaphore(MAX_CONCURRENT)
    tasks = [
        ("Task A", "Summarize recent AI news"),
        ("Task B", "Top Python libraries for data science?"),
        ("Task C", "Explain quantum computing simply"),
    ]
    async with httpx.AsyncClient(headers=auth) as client:
        results = await asyncio.gather(
            *[run_task(client, sem, n, p) for n, p in tasks]
        )
    for r in results:
        print(f"--- {r['task']} ---")
        print(r["answer"][:200])

asyncio.run(main())