Python

pip install codai-sdk

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)

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
)

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"))

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
)

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)

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)

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

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

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

# 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)

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"

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

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

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