# Python (/docs/sdks/python)

The codai-sdk package for Python 3.9+ — quickstart, every method, codai extensions and error handling.

`codai-sdk` is the official Python client. Zero dependencies (stdlib `urllib`), Python 3.9+, synchronous. The import name is `codai`.

```bash
pip install codai-sdk
```

<Callout type="warn">
The package is **`codai-sdk`**. `pip install codai` installs an unrelated project from another author that uses the same import name and shadows this SDK. If `from codai import Codai` fails, run `pip uninstall codai` and then `pip install codai-sdk`.
</Callout>

## Quickstart [#quickstart]

```python
import os
from codai import Codai

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

result = client.chat([{"role": "user", "content": "Explain asyncio.gather in one line."}])
print(result.content)
print(result.routed_to)  # upstream model that actually served

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

run = client.agents_run("Find and summarize the TODOs in this codebase")
print(run.result)

if result.request_id:
    client.feedback(result.request_id, 1)
```

## Configure [#configure]

```python
client = Codai(
    api_key=os.environ["CODAI_API_KEY"],  # required — raises ValueError if empty
    base_url="https://ai.codai.ro",        # default
    session_id="my-project",               # optional — session memory + sticky routing
    timeout=120.0,                         # seconds, default
    max_retries=2,                         # retries on 429 / 5xx with exponential backoff
)
```

## Resource groups [#resource-groups]

Since 0.2.0 every gateway operation is reachable as a method on the client, grouped by tag exactly like the [TypeScript SDK](./typescript.mdx) (`camelCase` becomes `snake_case`). The 0.1.x top-level methods below still work unchanged; `chat`, `embeddings`, `models` and `feedback` are callable resource groups, so `client.chat([...])` and `client.chat.completions.create({...})` are the same call.

| Attribute | Methods | Gateway |
| --- | --- | --- |
| `chat.completions` | `create`, `stream` | `POST /v1/chat/completions` |
| `messages` | `create`, `stream` | `POST /v1/messages` — Anthropic Messages wire |
| `responses` | `create`, `stream` | `POST /v1/responses` — OpenAI Responses wire |
| `embeddings` | `create` | `POST /v1/embeddings` |
| `audio` | `transcribe`, `transcribe_detailed`, `speech`, `speech_detailed` | `/v1/audio/*` |
| `tokens` | `create` | `POST /v1/tokens` |
| `models` | `list` | `GET /v1/models` |
| `health` | `get`, `ready`, `status` | `/health`, `/health/ready`, `/status` |
| `agents` | `run`; `runs.create`, `runs.get`, `runs.steps`, `runs.stats`, `runs.cancel`, `runs.stream` | `/v1/agents/*` |
| `tools` | `search`, `fetch`, `ground` | `/v1/tools/*` — [web tools](../agents/tools.mdx#web-tools-one-contract-everywhere) |
| `research` | `create`, `retrieve`, `wait`, `run` | `/v1/research` — [deep research](../api/gateway/research.mdx) |
| `vault` | `account`, `items`, `put_item`, `decide`, `catalog`, `use`, `http_call`, …; `devices.challenge`, `devices.enroll`, `devices.list`, `devices.rename`, `devices.revoke`; org vaults `members.list`, `members.candidates`, `members.add`, `members.update`, `members.remove`, `rotate_key`, `set_quorum`, `vote`; `simulate_policy`, `receipt_keys` | `/v1/vault/*` — [vault](../api/gateway/vault.mdx) |
| `cards` | `weather`, `places`, `geocode`, `static_map`, `fx`, `unfurl` | `/v1/cards/*` — [cards](../agents/cards.mdx) |
| `tasks` | `list`, `pending`, `stats`, `get`, `confirm` | `/v1/tasks/*` |
| `sessions` | `create`, `list`, `get`, `update`, `delete`, `dispatch`, `stream`; `events.*`, `controls.*`, `lease.*`, `shares.*` | `/v1/sessions/*` |
| `devices` | `list`, `update`, `delete`, `dispatch_inbox` | `/v1/devices/*` |
| `hosts` | `list`, `exec`, `post_result`, `stream` | `/v1/hosts/*` |
| `orgs` | `create`, `list`; `members.list`, `members.add`, `members.remove` | `/v1/orgs/*` |
| `account` | `get`, `update` | `/v1/account` |
| `receipt` | `get` | `GET /v1/receipt` |
| `feedback` | `submit` | `POST /v1/feedback` |
| `phone_models` | `list` | `GET /v1/phone/models` |

Every method takes an optional `ext={...}` bag covering all documented `X-Codai-*` request headers (`effort`, `thinking`, `thinking_budget`, `cache`, `no_task`, `task_id`, `incognito`, `mode`, `best_of`, `compact`, `device`, `share_token`, … plus raw `headers`) — see [headers](../gateway/headers.mdx). Request and response shapes are available as `TypedDict`s in `codai._types`, generated from the OpenAPI spec.

```python
client = Codai(api_key=os.environ["CODAI_API_KEY"], device="5dc0de00-0000-4000-8000-00000000c0da")

r = client.chat.completions.create(
    {"messages": messages, "model": "codai"},
    ext={"effort": "high", "thinking": True, "thinking_budget": 8192},
)

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

# codai-native SSE streams are generators of {"event", "data"} frames
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"])

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

## Chat [#chat]

```python
result = client.chat(
    messages=[{"role": "user", "content": "hello"}],
    model="codai",          # default
    temperature=None,
    max_tokens=None,
    tools=None,             # OpenAI function tools — you own the tool loop
    agent_mode=False,       # X-Codai-Mode: agent (Starter plan or higher)
    compact=None,           # "auto" → X-Codai-Compact: auto
    best_of=None,           # 3 forces best-of-N, 0 disables
    session_id=None,        # per-call override
)
```

`ChatResult` is a dataclass:

| Field | Type | Source |
| --- | --- | --- |
| `content` | `str` | First choice's assistant text. |
| `raw` | `dict` | The full OpenAI-shaped response. |
| `request_id` | `str \| None` | `x-request-id` — pass to `feedback()`. |
| `routed_to` | `str \| None` | `x-codai-routed-to`. |
| `exec_verify` | `str \| None` | `x-codai-exec-verify` on `codai-labs`. |
| `usage` | `dict \| None` | `{"prompt_tokens", "completion_tokens"}`. |

### Tool calls [#tool-calls]

```python
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
    },
}]

first = client.chat([{"role": "user", "content": "Weather in Cluj?"}], tools=tools)
message = first.raw["choices"][0]["message"]
for call in message.get("tool_calls") or []:
    args = json.loads(call["function"]["arguments"])
    output = get_weather(args["city"])
    final = client.chat([
        {"role": "user", "content": "Weather in Cluj?"},
        message,
        {"role": "tool", "tool_call_id": call["id"], "content": json.dumps(output)},
    ], tools=tools)
    print(final.content)
```

## Streaming [#streaming]

`chat_stream()` is a generator of **text deltas**. It parses `data:` frames, ignores keep-alives and stops on `[DONE]`.

```python
for delta in client.chat_stream(
    [{"role": "user", "content": "Write a haiku about Python."}],
    model="codai",
    temperature=None,
    max_tokens=None,
    agent_mode=False,
    session_id=None,
):
    print(delta, end="", flush=True)
```

`chat_stream()` yields text only. For the assembled tool calls, the final usage frame, the finish reason and the routing headers of a streamed reply use `client.chat.completions.stream()` and read `.final` after iterating (see [resource groups](#resource-groups)); `.chunks()` yields the raw `chat.completion.chunk` dicts. An HTTP error before the first byte raises `CodaiError`.

## Server-side agent [#server-side-agent]

```python
run = client.agents_run(
    task="Summarize the key points of the provided text.",  # required, ≤ 64 000 chars
    context="…your input…",                                  # optional, ≤ 256 000 chars
    system="Answer in Romanian.",                            # optional, ≤ 32 000 chars
    model=None,                                              # optional, default codai
)
run.result    # str
run.model     # upstream that served
run.event_id  # usage row id — pass to feedback()
run.usage     # dict | None
```

## Feedback [#feedback]

```python
client.feedback(request_id, 1)                    # 👍
client.feedback(request_id, -1, comment="wrong")  # 👎, comment ≤ 500 chars
```

The SDK sends `request_id` as `event_id`. Ratings feed the routing priors for everyone.

## Embeddings [#embeddings]

```python
vectors = client.embeddings(["hello", "world"], model="codai-embed", dimensions=None)
vectors[0]  # list[float]
```

Returns a list of vectors directly (not the raw envelope).

## Audio [#audio]

```python
# Speech-to-text — multipart upload, max 25 MB. Returns the transcript.
with open("clip.webm", "rb") as f:
    text = client.transcribe(f.read(), filename="clip.webm", model="codai-transcribe")

# Text-to-speech — returns audio bytes (mp3 by default).
audio = client.speech("Hello from codai.", model="codai-tts", voice="alloy")
with open("hello.mp3", "wb") as f:
    f.write(audio)
```

`transcribe()` does not retry — a multipart body cannot be replayed safely.

## Ephemeral tokens [#ephemeral-tokens]

```python
minted = client.mint_token("realtime", ttl_seconds=600)
minted["token"]       # "codai_eph_v1.…" — hand to a browser client
minted["expires_at"]  # ISO timestamp
minted["scope"]       # "realtime"
```

Scopes: `"realtime"`, `"audio"`, `"embeddings"`. TTL 60–3 600 s. See [ephemeral tokens](../gateway/embeddings-audio.mdx#ephemeral-tokens).

## Models [#models]

```python
for m in client.models():
    print(m["id"], m["codai"]["kind"])  # e.g. "codai alias"
```

Returns the `data` list of `GET /v1/models`.

## codai extensions [#codai-extensions]

| Option | Header sent | Effect |
| --- | --- | --- |
| `session_id` | `X-Codai-Session-Id` | Session memory, sticky routing, per-session receipts. Client-wide or per call. |
| `agent_mode=True` | `X-Codai-Mode: agent` | Plan-and-execute loop on the gateway (Starter plan or higher). |
| `compact="auto"` | `X-Codai-Compact: auto` | Deterministic server-side context compaction. `chat()` only. |
| `best_of=3` / `best_of=0` | `X-Codai-Best-Of` | Force or disable best-of-N sampling. `chat()` only. |

Everything else — effort tiers, thinking pins, task ids, device, share tokens — goes through the `ext={...}` bag accepted by every method (also by `chat()` and `chat_stream()`), with snake_case keys named after the header (`effort`, `thinking_pin`, `task_id`, …) and a raw `headers` dict as escape hatch; see [resource groups](#resource-groups) and the [headers](../gateway/headers.mdx) page.

## Errors [#errors]

```python
from codai import Codai, CodaiError

try:
    client.chat([{"role": "user", "content": "hi"}])
except CodaiError as e:
    print(e.status, e)          # HTTP status (0 for network failures after retries)
    code = (e.body or {}).get("error", {}).get("code") if isinstance(e.body, dict) else None
    if code == "quota_exceeded":
        ...  # plan task allowance used up — back off
```

`CodaiError.body` is the parsed JSON error envelope when the gateway returned one, otherwise `None` or the raw bytes. Stable `code` values are listed under [limits & pricing](../gateway/limits-pricing.mdx#error-contract).

## Async [#async]

There is no `asyncio` client in 0.1.1. The methods are blocking; in an async application wrap them with `asyncio.to_thread`, or call the gateway with an async HTTP client of your choice — the [wire formats](../gateway/wire-formats.mdx) are plain HTTP.

```python
result = await asyncio.to_thread(client.chat, [{"role": "user", "content": "hello"}])
```