Research runs

Ask

Send the question. The response comes back as soon as the run is admitted, not when it is answered — a run can take seconds or, with a compute job in it, hours — so what you get is a run with status: "running" and an id to follow.

curl https://corollary-labs.com/api/v1/runs \
  -H "Authorization: Bearer $COROLLARY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "input": "Fold this sequence and report its pLDDT: MKTAYIAKQRQISFVKSHFSRQ" }'

Pass thread_id to continue an earlier conversation — the agent then has everything said before, the files from earlier runs and the same sandbox. Leave it out and the run opens a new thread, whose id is thread_id on the run.

Follow

Two ways, and they can be mixed.

Poll. Read GET /runs/{id} untilstatus is no longer running. Every read returns what the run has written so far, so a progress display needs nothing else.

import time

while run["status"] in ("running", "cancelling"):
    time.sleep(3)
    run = requests.get(f"{BASE}/runs/{run['id']}", headers=H).json()

print(run["status"])
print(run["output"]["text"])

Stream. GET /runs/{id}/events is server-sent events: text as it is written, tools as they start and finish, and a final run event carrying the run exactly as the poll would return it. The connection ends on its own after a few minutes; if you have not seen run yet, reconnect with after set to the last event id you received and it resumes from there.

import json

def follow(run_id):
    after = -1
    while True:
        with requests.get(f"{BASE}/runs/{run_id}/events", headers=H,
                          params={"after": after}, stream=True) as r:
            event = None
            for raw in r.iter_lines(decode_unicode=True):
                if raw.startswith("id: "):     after = int(raw[4:])
                elif raw.startswith("event: "): event = raw[7:]
                elif raw.startswith("data: "):
                    data = json.loads(raw[6:])
                    if event == "text.delta":   print(data["text"], end="", flush=True)
                    elif event == "tool.result": print(f"\n[{data['tool']}: {data['status']}]")
                    elif event == "run":
                        if data["status"] != "running": return data

run = follow(run["id"])

Answer its questions

Before large or costly work the agent stops and asks: to approve a plan, to pick between routes, to allow a higher-quality image. In the app that is a card with buttons; over the API the run ends with status: "requires_action" and required_action says which tool is waiting and what it asked. Answer it and a new run carries on from the decision.

if run["status"] == "requires_action":
    action = run["required_action"]
    if action["tool"] == "confirm_scope":
        print("Plan:", *action["input"]["steps"], sep="\n  ")
        answer = {"approved": True}
    elif action["tool"] == "ask_choice":
        print("Options:", [o["id"] for o in action["input"]["options"]])
        answer = {"decision": "chosen", "chosen": [action["input"]["options"][0]["id"]]}
    else:  # request_quality_image
        answer = {"approved": False}

    run = requests.post(f"{BASE}/runs/{run['id']}/respond", headers=H,
                        json={"answer": answer}).json()
    run = follow(run["id"])     # a new run, with its own id

confirm_scope

A plan: input.objective, input.steps, sometimes input.cost and input.instead. Answer { "approved": true } to run it, false for the narrow answer instead.

ask_choice

A fork: input.question and input.options[] with ids. Answer { "decision": "chosen", "chosen": ["<id>"] } — several ids only when input.allowMultiple is true — or { "decision": "declined" }. An optional note is read by the agent.

request_quality_image

Permission for a costlier image: input.prompt and input.reason. Answer { "approved": true | false }.

Read the results

A finished run has output.text, the answer as Markdown, and output.tool_results, one entry per tool the agent used. Each has the same envelope whatever the tool: which tool, whether it succeeded, what it was asked, what it returned, and its files.

{
  "type": "tool_result",
  "call_id": "call_01a0…",
  "tool": "fold_protein",
  "status": "succeeded",
  "input": { "sequence": "MKTAYIAKQRQISFVKSHFSRQ", "label": "query" },
  "output": { "ok": true, "residues": 22, "meanPlddt": 0.62, "structureId": "art_k3m9x2q7", "…": "…" },
  "files": [
    { "id": "art_k3m9x2q7", "role": "structureId", "url": "https://corollary-labs.com/api/v1/files/art_k3m9x2q7" }
  ]
}

outputis the tool's own payload and changes when the tool improves; the envelope around it does not. Build on tool, status, input and files, and readoutput for the tool you asked for. See the runs reference for every field.

Fetch the files

Anything heavy — coordinates, figures, tables, pages, meshes — is a file, not inline JSON. Every files[].url serves the bytes with their media type when called with your key.

for result in run["output"]["tool_results"]:
    for f in result["files"]:
        r = requests.get(f["url"], headers=H)
        name = f.get("name") or f["id"]
        open(name, "wb").write(r.content)     # e.g. a PDB, a PNG, a CSV

• Files are kept for thirty days. After that the URL is a 404; run the tool again to regenerate.
• A file belongs to your account. Another account's file id is simply not found.

Retrying safely

Send an Idempotency-Key header with POST /runs or /respond and a retried request — a timeout, a crashed worker — returns the run the first attempt started instead of starting another. Any string up to 200 characters; a UUID is the usual choice. A thread that already has a run going answers 409 thread_busy to a new one: follow or cancel the first.