codai-sdk — Python SDK

Official Python client for the codai AI gateway. Zero dependencies (stdlib only), Python 3.9+.

pip install codai-sdk

The package is codai-sdk. pip install codai installs an unrelated project from another author that uses the same import name and will shadow this SDK. If from codai import Codai fails, run pip uninstall codai and pip install codai-sdk.

You need a codai API key — keys start with codai_. Create one at hub.codai.ro and export it as CODAI_API_KEY. The import name is codai:

import os

from codai import Codai

client = Codai(api_key=os.environ["CODAI_API_KEY"], session_id="my-project")

# Chat (OpenAI-compatible, smart-routed)
result = client.chat([{"role": "user", "content": "Explain asyncio.gather"}])
print(result.content)
print(result.routed_to)   # which model actually served

# Streaming
for delta in client.chat_stream([{"role": "user", "content": "hi"}]):
    print(delta, end="", flush=True)

# Server-side agent loop
run = client.agents_run("Find and summarize the TODOs in this codebase")
print(run.result)

# Feedback (improves routing for everyone)
client.feedback(result.request_id, 1)

codai extensions

Resource groups (0.2.0)

Every gateway operation is a method on the client, grouped exactly like the TypeScript SDK (camelCase → snake_case). The 0.1.x helpers above still work — client.chat([...]) and client.chat.completions.create({...}) are the same call. Every method takes an optional ext={...} bag of X-Codai-* headers (effort, thinking, thinking_budget, cache, no_task, task_id, device, compact, best_of, share_token, … plus raw headers).

from codai import Codai, CodaiError

client = Codai(api_key="codai_...", device="5dc0de00-0000-4000-8000-00000000c0da")

# Chat with extension headers; raw chunks + aggregate on streams
r = client.chat.completions.create(
    {"model": "codai", "messages": [{"role": "user", "content": "hi"}]},
    ext={"effort": "high", "thinking": True, "thinking_budget": 8192},
)
print(r.content, r.routed_to, r.usage)

stream = client.chat.completions.stream({"messages": [{"role": "user", "content": "hi"}]})
for delta in stream:            # text deltas; stream.chunks() for raw chunk dicts
    print(delta, end="")
print(stream.final.usage, stream.final.tool_calls)

# Anthropic / OpenAI Responses wires
client.messages.create({"messages": [{"role": "user", "content": "hi"}], "max_tokens": 200}).text
client.responses.create({"input": "hi"}).output_text

# Async agent run + SSE progress
run = client.agents.runs.create({"task": "Summarise the repo README."})
for ev in client.agents.runs.stream(run["id"]):
    if ev["event"] == "done":
        print(ev["data"]["status"], ev["data"]["result"])

# Tasks, account, receipt
page = client.tasks.list(limit=20)
print(client.account.get()["plan"], client.receipt.get(since="2026-09-01T00:00:00Z"))

# Shared sessions (needs a device id on the client) and hosts
s = client.sessions.create({"title": "pairing"})
for ev in client.sessions.stream(s["id"], after=0):
    print(ev["event"], ev["data"].get("seq"))
print(client.hosts.list())

try:
    client.orgs.members.add("org-1", {"user_id": "u", "role": "member"})
except CodaiError as e:
    print(e.status, e.code, e.request_id, e.retry_after)

Typed request/response shapes live in codai._types (generated from the OpenAPI spec — TypedDicts such as ChatCompletionRequest, Task, SessionEvent, CreateSessionShareResponse).

Lists and the share sheet

client.lists covers /v1/lists/* and POST /v1/share: list, create, reorder, update, delete, items, add, update_item, delete_item, image (ListItemImage with data + content_type), export_markdown (str), share and stream (live list events). add and share are idempotent on client_id / the canonical URL: a duplicate answers 200 with created: False.

shared = client.lists.share({"url": "https://example.com", "auto": True})
md = client.lists.export_markdown(shared["list"]["id"])
for ev in client.lists.stream():
    print(ev["op"], ev.get("listId"), ev.get("itemId"))

client.schedules.audit(schedule_id=None, limit=None) returns the scheduler audit log (GET /v1/schedules/audit), newest first.

codai Resolve (0.4.0)

Pay per verified fix: submit a public GitHub issue, get a flat quote, accept it (held on your wallet), and pay only when the fix passes your tests. First 3 fixes are free. Lives on https://resolve.codai.ro (same key; resolve_base_url= to override).

from codai import Codai, ResolveInsufficientFundsError

client = Codai(api_key="codai_...")
job = client.resolve.submit({
    "repo_url": "https://github.com/owner/repo",
    "issue_url": "https://github.com/owner/repo/issues/42",
})
if job["status"] == "quoted":
    try:
        client.resolve.accept(job["id"])
    except ResolveInsufficientFundsError as e:
        print("top up first:", e.topup_url, e.bundle_url or "")  # bundle_url: t7 only
        raise
done = client.resolve.wait_for(job["id"])
print(done["status"], done.get("attestation_url"))

Failed attempts and infrastructure errors refund the hold in full. Types: codai._resolve_types (generated from resolve.yaml).

Pricing (0.7.0). Your first paid accept holds €1 instead of the list price (the quote then carries intro_price_micro_eur / list_price_micro_eur); it is used up only by a verified fix — a failed intro job is refunded and the intro is offered again. A fix bundle (€49 = 10 credits, https://pay.codai.ro/checkout?bundle=resolve10) pays for t7 jobs only: a credit is taken at accept instead of a wallet hold and returned on failure. Precedence: free > intro > credit (t7) > wallet; accept() reports it in paid_with, a t7 402 sets bundle_url, and client.resolve.balance() returns fix_credits, intro_available and wallet_micro_eur.

Webhooks instead of polling (0.5.0)

Pass callback_url (https, public host) on submit; the response carries callback_secret once — store it. Resolve POSTs job.quoted, job.declined, job.resolved, job.failed and job.error, signed with x-codai-signature: t=<unix>,v1=<hex HMAC-SHA256>. Verify the raw body with verify_resolve_webhook (constant-time, 300 s tolerance by default). Up to 3 attempts on 5xx/429/network errors; deduplicate on x-codai-delivery; delivery never changes job state — get(id) stays the source of truth.

import json
from codai import verify_resolve_webhook

job = client.resolve.submit({
    "repo_url": "https://github.com/owner/repo",
    "issue_url": "https://github.com/owner/repo/issues/42",
    "callback_url": "https://ci.example.com/hooks/codai-resolve",
})
save_secret(job["id"], job["callback_secret"])  # shown once

# in your webhook handler (FastAPI: raw = await request.body())
if not verify_resolve_webhook(secret, raw, request.headers.get("x-codai-signature")):
    raise HTTPException(401, "bad signature")
event = json.loads(raw)  # event["event"], event["job"]["attestation_url"]

Verifying a signed attestation (0.6.0)

Attestations created from 2026-09-27 carry signature: a DSSE envelope (ECDSA P-256/SHA-256) over a canonical statement of sha256 hashes of the patch, repro test and test outputs, plus the commands, base commit, runner digest and timestamps. verify_attestation checks that the signed statement matches the fields you were served (stdlib) and the signature against GET /v1/resolve/keys — the signature step needs pip install cryptography (the SDK stays dependency-free; without it an ImportError explains what to install). Older attestations have signature: None → {"ok": False, "reason": "UNSIGNED"}.

proof = client.resolve.attestation(job["attestation_url"])
r = client.resolve.verify_attestation(proof)  # or keys=pinned_keys
if not r["ok"]:
    raise SystemExit(f"not verified: {r['reason']}")  # BAD_SIGNATURE | STATEMENT_MISMATCH | ...
print("signed by", r["keyid"])

Development

python -m venv .venv && .venv/Scripts/python -m pip install -e ".[dev]"
.venv/Scripts/python -m pytest -q             # unit + OpenAPI parity tests, no network
.venv/Scripts/python scripts/gen-types.py     # regenerate src/codai/_types.py

MIT © codai