Metadata-Version: 2.4
Name: cosmo-ai-sdk
Version: 0.2.0
Summary: Cosmo Realtime SDK — typed event-stream client for voice + multimodal sessions
Author: Socratic Inc.
License: Apache-2.0
Project-URL: Homepage, https://askcosmo.ai
Project-URL: Repository, https://github.com/socratic-ai/cosmo-python-sdk
Keywords: cosmo,realtime,voice,livekit,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27
Requires-Dist: livekit>=0.18
Requires-Dist: pydantic>=2.7
Requires-Dist: sounddevice>=0.4
Requires-Dist: structlog>=23.1
Provides-Extra: livekit
Provides-Extra: audio
Provides-Extra: mcp
Requires-Dist: mcp<2,>=1.14; extra == "mcp"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Dynamic: license-file

# cosmo-ai-sdk (Python)

Async SDK for the Cosmo Realtime external API.
One call starts a session (REST session-start + LiveKit room join); the
session is an async iterator of typed events. Audio I/O rides the LiveKit
room at the platform layer.

> New to the SDK? Start with the [documentation](https://platform.askcosmo.ai/docs)
> — getting started, the credential model, and the expected session lifecycle.

---

## Install

```bash
pip install cosmo-ai-sdk
```

> **Beta.** The SDK is pre-1.0: minor releases may include breaking API
> changes, called out in the
> [changelog](https://platform.askcosmo.ai/docs/meta/changelog). We will tag 1.0 once the session
> event stream, tool authoring, and credential APIs have gone a full release
> cycle without breaking changes. Pin a minor version (e.g.
> `cosmo-ai-sdk~=0.2.0`) if you need stability.

Requires Python 3.10+. One install covers everything: the media transport
(`livekit>=0.18`) and OS audio I/O (`sounddevice` — microphone capture via
`set_microphone_enabled`, speaker playback via `set_speaker_enabled`) are
included. Two platform notes: `livekit` is a native wheel published for
macOS, glibc Linux (x86_64/aarch64), and Windows x64 — on musl-based images
(e.g. Alpine) there is no wheel, so use a `python:*-slim` base instead. And
on Linux, using the OS mic/speaker needs the system PortAudio library
(`apt install libportaudio2`); servers that never touch OS audio don't need
it.

---

## Quickstart

Three objects, one per concern of running a session:

- **`CosmoRealtime`** — the connection: credential, endpoint, HTTP transport.
- **`Agent`** — the persona/configuration of the model (instructions, model,
  voice, tools, turn-taking), reusable across sessions.
- **session** (`agent.start(...)`) — one live run plus its per-run,
  transport-level options (resume, recording opt-out, lifecycle observer).

```python
import asyncio
from cosmo_ai import (
    CosmoRealtime,
    RealtimeReady,
    RealtimeSessionEnded,
    RealtimeTranscriptDelta,
    RealtimeTranscriptRole,
    RealtimeUsage,
)

async def main() -> None:
    client = CosmoRealtime(api_key="cosmo_...")
    agent = client.agent(
        instructions="You are a terse assistant.",
        voice="Upbeat",
    )
    async with agent.start() as session:
        await session.send_text("Hello!")

        async for event in session:
            match event:
                case RealtimeReady():
                    print(f"ready — session {event.session_id}")
                case RealtimeTranscriptDelta(is_final=True):
                    # Compare against the enum, not a bare string.
                    who = "agent" if event.role is RealtimeTranscriptRole.ASSISTANT else "you"
                    print(f"[{who}] {event.text}")
                case RealtimeUsage():
                    print(f"tokens so far: {event.total_tokens}")
                case RealtimeSessionEnded():
                    print(f"ended: {event.reason}")
                    break

asyncio.run(main())
```

The stream stays open until the session ends, so `break` on the event you were
waiting for — otherwise the loop simply keeps waiting.

See `hello_realtime.py` in the [examples repo](https://github.com/socratic-ai/cosmo-examples/tree/main/python) for the runnable
text-only demo and `voice_cli/` there for a live two-way voice call
(microphone in, agent audio out).

The client talks to `https://platform.askcosmo.ai` by default. For local
development against another backend, set the `COSMO_BASE_URL` environment
variable (`http://` is allowed only for localhost).

---

## Credentials & end-user tokens

Construct the client with exactly one credential:

- **`api_key`** — workspace-scoped, **secret, server-side only**. Can mint
  end-user tokens and open sessions.
- **`token`** — a minted end-user JWT scoped to one external user. Safe to
  hand to a browser/device; can open sessions but cannot mint.

For multi-user apps, mint on your backend and connect on the client:

```python
# 1) Backend (api key) — mint a short-lived token for an end user
backend = CosmoRealtime(api_key="cosmo_...")
minted = await backend.mint_token("user-123")   # -> {jwt, expires_at}; send minted.jwt to the user

# 2) Browser / device (minted token) — connect as that user
client = CosmoRealtime(token=minted.jwt)
async with client.agent(instructions="...").start() as session:
    async for event in session:
        ...
```

`mint_token` is idempotent per `(workspace, external_user_id)` — the same
external user maps to the same auto-provisioned project. The `api_key` never
leaves your server; the browser only holds the short-lived, per-user JWT.

### Verifying a credential

`verify()` checks a credential without starting a session — no room, no agent,
no charge. Use it as a startup check or a CI smoke test:

```python
info = await client.verify()          # raises VerifyError if the credential is bad
info.workspace                        # which workspace (and so which environment)
info.scopes                           # e.g. ["realtime:use"]
info.can_start_sessions               # False -> valid credential, missing realtime:use
info.realtime_voice_available         # False -> no default voice stack configured here
```

It works with either credential; a minted token also reports the
`external_user_id` it is bound to, and gets `workspace=None` — it runs on an
end user's device, which is not told whose workspace it belongs to. An
under-scoped credential is a returned fact, not an exception — only a
credential the server rejects raises.

For a private-CA / self-signed https backend (or proxies, mTLS, custom
transport), point the SDK at it with `COSMO_BASE_URL` and supply your own
`httpx.AsyncClient` via `http_client` — it controls TLS/transport. An injected
client is yours: the SDK uses it but never closes it on `aclose()`, and it
applies its own timeout to session-start/mint requests so your client's timeout
won't shorten them.

```python
import httpx
# COSMO_BASE_URL=https://internal.example
client = CosmoRealtime(
    api_key="cosmo_...",
    http_client=httpx.AsyncClient(verify="/etc/ssl/corp-ca.pem"),
)
```

---

## Consumption model

`async for event in session` is **the** way to observe a session. The stream
yields the `RealtimeSessionEvent` union — `ready`, `transcript`,
`model-text`, `turn-complete`, speaking/LLM/TTS phase events, the tool
lifecycle (`tool-call`, `tool-dispatch-started`, `tool-result`,
`tool-invocation`), `reconnecting`, `error`, `pong`, `cosmo.usage` — all
Pydantic models discriminated by `type`.

Transcript roles are `RealtimeTranscriptRole.USER` / `.ASSISTANT`, whose values
are lowercase (`"user"` / `"assistant"`). The wire spells them uppercase and
decoding accepts either casing, so compare against the enum member rather than
a string literal.

`RealtimeUsage` (`cosmo.usage`) carries **cumulative** token counts for the
session, split by direction and modality — each event supersedes the last. A
provider that reports no usage emits none, so absence is not zero.

Two guarantees:

* **Unknown ≠ fatal.** A frame with an unrecognized `type` (or one that
  fails validation) surfaces as `UnknownEvent(raw_type=...)` and the stream
  continues. Undecodable frames surface as `UnknownEvent(raw_type=None)`.
* **`RealtimeSessionEnded` is always the final item.** An SDK-local terminal
  sentinel synthesized on `session.end()` / context-manager exit / transport
  close. The server's own best-effort `session-ended` frame never surfaces
  mid-stream — its reason is latched and carried by the terminal sentinel
  (with a short grace teardown if the room close never follows). After it,
  iteration finishes.

Oversized server messages arrive as `server-envelope-chunk` frames; the SDK
reassembles them transparently — chunks never surface as events.

## Logging

The SDK logs nothing by default. Its loggers sit under the `cosmo_ai` stdlib
namespace behind a `NullHandler`, so diagnostics never interleave with your
own output until you ask for them:

```python
import logging

logging.basicConfig()
logging.getLogger("cosmo_ai").setLevel(logging.DEBUG)
```

Records carry structured key/value context (the SDK uses `structlog`), and an
app that configures `structlog` gets them rendered in its own format.

## Public API

### `CosmoRealtime`

| Member | Description |
|---|---|
| `CosmoRealtime(api_key=... \| token=..., http_client=None)` | Construct with exactly one credential (see [Credentials](#credentials--end-user-tokens)). The base URL defaults to `https://platform.askcosmo.ai`; override it with the `COSMO_BASE_URL` environment variable for local development. `http_client` injects your own `httpx.AsyncClient` (custom CA/TLS, proxies, mTLS, transport) — the SDK uses but never closes it. Reuse across sessions; close the owned client with `aclose()` / `async with` |
| `agent(*, instructions=None, model=None, model_options=None, voice=None, tools=None, interruption_sensitivity=None, greeting=None, audio=None, mcp=None, skills=None, hooks=None)` | Build a reusable inline `Agent` — the persona (see [`Agent`](#agent)), including its opening `greeting` and its `audio` pipeline (`AudioConfig`: `output`, `noise_cancellation`, `ambience`). `voice` is the voice id as a plain string, or a `VoiceConfig(name=..., speaking_style=...)`. `tools` is a list of `ClientTool` / typed server opt-ins (`WebSearchTool`, `ExamineImageTool`, `DetectObjectsTool`, `PointAtObjectTool`); `mcp` / `skills` / `hooks` attach the concepts described in their sections below. Fields left `None` fall back to the server default, except `audio.noise_cancellation`, which the SDK defaults to `True` — background voices are cancelled out of the user's inbound audio before it reaches the agent, and since the denoised signal is also what turn-taking reads, pass `AudioConfig(noise_cancellation=False)` to opt out and keep the raw microphone signal |
| `catalog_agent(name, *, inputs=None, voice=None, tools=None, mcp=None, hooks=None)` | Build an `Agent` that runs a workspace catalog agent by machine handle; the stored config runs verbatim. `inputs` fills the agent's declared input fields; `voice` (string or `VoiceConfig`) overrides the stored voice for this run only; `tools` / `mcp` add client-executed declarations. There are no other persona parameters — sending stored config with a catalog launch is a type error |
| `mint_token(external_user_id)` | Mint a short-lived end-user JWT (`-> MintedToken{jwt, expires_at}`). Requires an `api_key` credential; raises `MintTokenError` otherwise |
| `verify()` | Check the credential without starting a session (`-> CredentialInfo`). Free — see [Verifying a credential](#verifying-a-credential). Raises `VerifyError` if the credential is rejected |
| `aclose()` | Close the owned HTTP client (also an async context manager) |

### `Agent`

A reusable persona built by `client.agent(...)`; one agent opens any number of
sessions.

| Member | Description |
|---|---|
| `start(*, resume_session_id=None, store_recording=None, on_state_change=None)` | Open a session: POST `session/start` (a `session-config` payload) + LiveKit join. These are the per-run, transport-level options (resume, recording opt-out, lifecycle observer); persona fields — including `greeting` and the `audio` pipeline — ride unchanged from the agent (build another agent to change them). Returns a handle that is an **async context manager** (`async with agent.start() as session:` — ends the session on exit) and is also awaitable (`session = await agent.start()` if you own the lifecycle). Raises `VersionMismatchError` when the server refuses the protocol version, `SessionStartError` for any other rejection |

### `RealtimeSession`

| Member | Description |
|---|---|
| `async for event in session` | The typed event stream (see above) |
| `send_text(content)` | Send a text turn the agent answers. For a session that never speaks, configure the agent with `audio=AudioConfig(output=False)` |
| `send_context(content)` | Give the agent context without asking it anything: no turn, no speech, no transcript entry. For live application state |
| `set_muted(muted)` | Toggle the server-side mic gate |
| `ping()` | Heartbeat; server replies with a `RealtimePong` event |
| `activity_end()` | Manual end-of-turn (wake-word gating only) |
| `send_image(data=..., mime_type=..., stream_id=...)` | One base64 image frame |
| `end()` | Graceful end: `end` frame + finish the stream + leave the room |
| `close()` | Abrupt local teardown without telling the server |
| `set_microphone_enabled(enabled)` | Capture + publish the default OS mic (or stop it), gating the server side (livekit-rtc Python has no native mic capture; the SDK supplies it via `sounddevice`). See `voice_cli` in the [examples repo](https://github.com/socratic-ai/cosmo-examples/tree/main/python) |
| `set_speaker_enabled(enabled)` | Play the agent's voice on the default OS output device (or stop it) (livekit-rtc Python has no native playback; the SDK supplies it via `sounddevice`). See `voice_cli` in the [examples repo](https://github.com/socratic-ai/cosmo-examples/tree/main/python) |
| `set_agent_playback_volume(volume)` | Software gain 0…1 for OS playback (clamped; `0` mutes). Affects only `set_speaker_enabled` output, never `agent_audio()` frames; may be set before the speaker is enabled |
| `agent_audio()` | Async-iterate the agent's decoded voice as 16-bit mono PCM `AgentAudioFrame`s (`cosmo_ai.audio`) — record it, pipe it to telephony, or feed a custom player. Multiple concurrent iterators each get every frame; a stalled consumer drops its oldest. Finishes when the session ends |
| `audio_levels()` | Async-iterate `AudioLevels(mic, agent)` (`cosmo_ai.audio`) RMS levels (0…1) at ~20 Hz, latest-value. Iterating activates agent-audio decode; fields read `0.0` while their source is inactive |
| `publish_audio_source(source)` | Advanced: publish a caller-owned `rtc.AudioSource` and keep feeding it frames yourself (synthetic generator, WAV replay, load tests) |
| `dial(phone_number)` | Place an outbound phone call into this session — the callee joins as a SIP participant (see [Outbound calling](#outbound-calling)) |
| `start_screen_share()` / `push_screen_share_frame(frame)` / `stop_screen_share()` | Screen-share publish |
| `add_video_stream()` / `handle.push(frame)` / `remove_video_stream(handle)` | Publish video that is **not** the user's screen — a camera, a file, any pixels-only source (see [Publishing video](#publishing-video)) |
| `state` / `session_id` / `response` / `config` | Lifecycle snapshot + start results |

### Tools

Define a client-executed tool with the `@tool` decorator: the first
parameter's Pydantic model drives the model-facing JSON Schema, runtime
validation, and the typed arguments the handler receives.

```python
from typing import Any, Literal

from pydantic import BaseModel, Field

from cosmo_ai import WebSearchTool, tool


class WeatherInput(BaseModel):
    city: str = Field(description="City name")
    unit: Literal["c", "f"] = "c"


@tool  # or @tool(name=..., description=..., background=False)
async def get_weather(input: WeatherInput) -> dict[str, Any]:
    """Current weather for a city."""
    return {"temp_c": await lookup(input.city)}


agent = client.agent(
    tools=[get_weather, WebSearchTool()],
)
async with agent.start() as session:
    async for event in session:
        ...
```

- **Name** defaults to the function name, **description** to the docstring
  (both overridable via `@tool(name=..., description=...)`; the description is
  model-facing and required).
- **Everything checks at decoration**, not at connect: signature shape, name
  and description limits, and the emitted schema against the backend's
  restricted JSON-Schema dialect. A model whose schema the server would
  reject (regex `pattern`s, `format`s, recursive models, `extra="forbid"`, …)
  raises `ToolSchemaError` at import/startup instead of surfacing in
  `RealtimeReady.rejected_tools` mid-connect.
- **Malformed model calls never reach your code.** Arguments are validated
  with `Model.model_validate` first; a failure becomes a sanitized
  `INVALID_INPUT` tool error the model can self-correct from (paths +
  constraints only — submitted values never appear in the error or in logs).
- **Validation semantics are Pydantic's**: coercion applies, defaults are
  filled in. The emitted schema describes the accepted input; the handler
  receives the validator's output.
- **Long-running work**: `@tool(background=True)` expects
  `async def fn(input: Model, job: ClientToolJob)` and follows the unchanged
  job contract (`job.ack(...)`, then `job.complete(...)` / `job.fail(...)`).

The decorator lowers to a plain `ClientTool`, which stays public in
`cosmo_ai.tools` as the advanced escape hatch — hand-write one to
control the raw JSON Schema yourself (its handler then receives an
unvalidated `dict`):

```python
from cosmo_ai.tools import ClientTool

ClientTool(
    name="get_local_time",
    description="Returns the local wall-clock time.",
    parameters={"type": "object", "properties": {}, "required": []},
    handler=get_local_time,  # async (args: dict) -> dict
)
```

When the agent invokes a client tool the SDK calls the handler and reports
the returned dict back as the result; raise to surface a tool error. The
handler is local-only: it is excluded from serialization and never crosses
the wire. Every client tool carries a handler — constructing a spec without
one is a validation error, since a declared tool the client cannot execute
would fail on every invocation.

Typed opt-in classes enable built-in server-executed tools — `WebSearchTool`
(web search), `ExamineImageTool` (full-resolution frame examination),
`DetectObjectsTool` / `PointAtObjectTool` (object locators). Each is zero-config: the server
owns the model-facing declaration; you only opt in.

Client-tool specs the server refuses are echoed on
`RealtimeReady.rejected_tools`; the session still starts without them.

The `cosmo_sdk_` name prefix is **reserved for the client tools the SDK ships
itself** — the SDK owns their names and schemas, so a caller's tool taking one
would swap it for something the model was told behaves differently. A tool of
your own carrying the prefix is rejected where you declared it rather than at
connect (the wider `cosmo_` namespace belongs to server tools, which the server
rejects too); every other name stays free, including the natural ones the SDK's
own tools shorten to.

### Publishing video

`start_screen_share` publishes the user's screen. `add_video_stream` publishes
everything else — a camera, a decoded file, a rendered scene:

```python
from livekit import rtc

stream = await session.add_video_stream()
while capturing:
    pixels = grab_frame()                      # yours: OpenCV, picamera, a file
    stream.push(rtc.VideoFrame(width, height, rtc.VideoBufferType.RGB24, pixels))
await session.remove_video_stream(stream)
```

The two publish under different LiveKit sources, which is how the backend
tells them apart: a camera feed is described to the model as one, and the
screen tools stay anchored to actual screen shares. **One video publish at a
time** — a second `add_video_stream`, or a `start_screen_share` while a stream
is live, raises `VideoPublishAlreadyActiveError` (`from cosmo_ai import
VideoPublishAlreadyActiveError`).

`push` is safe to call from whatever thread your capture loop runs on; the
publish it triggers is handed to the session's event loop.

Capturing frames is yours. The SDK opens no camera and brings no dependency
for one; anything that can produce an `rtc.VideoFrame` works, and LiveKit
accepts the common buffer layouts (`RGB24`, `RGBA`, `BGRA`, `I420`, …) so
there is usually no conversion to write. The publish is deferred to the first
frame, so dimensions resolve from the source rather than from the arguments.

### Drawing on the user's live view

`draw_box` / `draw_point` are the renderer half of the locate-then-draw pair.
A locator (`DetectObjectsTool` / `PointAtObjectTool`) returns candidate boxes
or points to the model; the model picks the one matching what it is looking at
and passes it to the renderer, which draws it over the user's live camera or
screen preview. You supply one function of request → outcome; the SDK owns the
name, description, schema, decode, and reply shape.

```python
from cosmo_ai import DetectObjectsTool
from cosmo_ai.tools import DrawBoxRequest, DrawOutcome, draw_box


def on_draw(request: DrawBoxRequest) -> DrawOutcome:   # sync or async
    if not camera.streaming:
        return DrawOutcome(
            shown=False,
            reason="the camera is off — ask the user to turn it on",
        )
    overlay.show(request.box, label=request.label)
    return DrawOutcome(shown=True)


agent = client.agent(tools=[DetectObjectsTool(), draw_box(on_draw)])
```

- **Coordinates are normalized** to the frame the model was shown — `[0,1]`,
  top-left origin — so you map them onto the preview the same way you map a
  locator's box. Out-of-range values are clamped, so a model that overshoots
  the frame edge still yields a drawable annotation.
- **Malformed arguments never reach your handler**; they surface to the model
  as the invocation's error.
- **Answer honestly.** `DrawOutcome`'s `reason` is model-facing prose the agent
  says out loud, not an error code — a box reported as shown but invisible
  leaves the model talking about something the user cannot see.

`draw_point` follows the same contract with a `DrawPointRequest`. The two exist
side by side because they answer different questions: a box around a leaf
includes everything behind it, where a marked point says one thing.

If your app shows the frames it publishes, `box_rect` / `point_position`
(`cosmo_ai.tools`) put the annotation where the model pointed: they map a
normalized box or point onto the view showing the frame, correcting for the
crop or letterbox and for a mirrored selfie preview. Pure arithmetic, in
whatever coordinate space your preview uses — the SDK owns no window and draws
nothing.

```python
from cosmo_ai.tools import Size, box_rect

rect = box_rect(
    request.box,
    container=Size(view_width, view_height),
    frame_size=Size(frame_width, frame_height),
    content_mode="fill",
    mirrored=front_camera,
)
```

### Outbound calling

`session.dial(phone_number)` places an outbound phone call **into a running
session**: the dialed party joins the session's room as a SIP participant and
the agent — already in the room — converses with them. `start()` stays about
creating the session; choosing participants (the local mic, or a phone callee)
is always a separate, explicit step.

```python
async with agent.start() as session:        # session only — no participants yet
    await session.dial("+14155550199")        # bring the callee in over SIP
    async for event in session:               # transcripts, etc., as usual
        ...
```

- **Number format** — E.164 (`+` then 8–15 digits); the SDK fast-fails a
  malformed number with `DialError(code="invalid_phone_number")` before any
  request.
- **Enablement** — outbound calling must be enabled for the workspace
  (`phone_calls_disabled` otherwise); the same `realtime:use` credential that
  opened the session authorizes the dial.
- **Limits** — calls count against the workspace's weekly per-user minute limit.
- **Errors** — server rejections raise `DialError` with the server's slug
  (`phone_calls_disabled`, `minute_limit_exceeded`, `session_not_live`,
  `forbidden`, …).
- **Return** — `DialResult{dial_id}`, a handle to the queued call. The call
  rings asynchronously; watch `session` events for the conversation.

The call is a transport-level REST request (not a data-channel send), so it is
the one `RealtimeSession` method that reaches the API directly. See
`outbound_call.py` in the [examples repo](https://github.com/socratic-ai/cosmo-examples/tree/main/python).

---

## MCP servers (local stdio)

Expose any local [MCP](https://modelcontextprotocol.io) server's tools to the
realtime model. Attach servers with the `mcp` argument — a Claude-Code
`.mcp.json` config file (one file describes many servers), or a list mixing
config files and inline `McpStdioServer` objects (a path element expands in
place). The SDK spawns each server at session start, lists its tools, and
proxies calls — tools are namespaced `mcp__<server>__<tool>`.

```python
from cosmo_ai import CosmoRealtime

client = CosmoRealtime(api_key="cosmo_...")
agent = client.agent(instructions="You are Alex.", mcp="./mcp.json")
```

```python
from cosmo_ai.mcp import McpStdioServer

agent = client.agent(
    instructions="You are Alex.",
    mcp=[McpStdioServer(name="fs", command="npx", args=("-y", "@modelcontextprotocol/server-filesystem", "/tmp"))],
)
```

A missing or malformed config file and duplicate server names raise
`McpConfigError` when the agent is built, not mid-call. Requires the `mcp`
extra: `pip install 'cosmo-ai-sdk[mcp]'` (a live connect without it raises
`McpExtraNotInstalled`). v1 supports **stdio** servers; remote (`http`/`sse`)
entries in `.mcp.json` are skipped with a warning so the file stays shareable
with harnesses that support them. An `McpStdioServer` runs an arbitrary local
command — trust your config.

---

## Hooks

Attach in-process callbacks at the session's four lifecycle seams —
`SessionStart`, `PreToolUse`, `PostToolUse`, `SessionEnd`. Observe
everything; override the two seams
the client controls (inject start-of-session context; deny or rewrite a local
client-tool call).

Declare a hook with the seam's decorator — the decorated name becomes a
`Hook` — and attach with `hooks=[...]`; list order is fold order:

```python
from cosmo_ai import CosmoRealtime, hooks
from cosmo_ai.hooks import PreToolUseResult

@hooks.pre_tool_use(matcher="delete_*")
def block_deletes(ctx) -> PreToolUseResult:
    return PreToolUseResult(permission="deny", reason="destructive tools are disabled")

@hooks.session_end
async def log_end(ctx) -> None:
    print("session ended:", ctx.reason.value)

client = CosmoRealtime(api_key="cosmo_...")
agent = client.agent(instructions="You are Alex.", hooks=[block_deletes, log_end])
```

The same list also carries **server hooks** — declarative rules the server
executes (they work even if your process dies mid-call): `SilenceTimeout` with
a `Say` or `EndCall` action, e.g.
`hooks=[log_end, SilenceTimeout(timeout_seconds=45, action=EndCall())]`.
A fired server hook reaches you as a `RealtimeUserSpeechTimeout` event on the
session's event stream, not as a hook.

A throwing hook is logged and skipped — it never breaks the session. A
malformed matcher raises at decoration, not at session start. `SessionStart`
`additional_context` is appended to the instructions; `PreToolUse`
`deny`/`updated_arguments` apply only to locally-executed client tools.

See `hooks_agent.py` in the [examples repo](https://github.com/socratic-ai/cosmo-examples/tree/main/python) for a runnable end-to-end example.

---

## Package layout

| Module | Contents |
|---|---|
| `cosmo_ai.client` | `CosmoRealtime` |
| `cosmo_ai.session` | `RealtimeSession`, `SessionState`, `DisconnectReason` |
| `cosmo_ai.tools` | `tool` (also re-exported at the root), the renderer tools the SDK ships (`draw_box`, `draw_point`, `DrawBoxRequest`, `DrawPointRequest`, `DrawOutcome`, `NormalizedBox`, `NormalizedPoint`), plus advanced authoring: `ClientTool`, `BackgroundClientTool`, `ClientToolJob`, `ToolSchemaError` |
| `cosmo_ai.skills` | `Skill`, `SkillParseError`, `SkillsInput` — the `skills=` argument takes a directory or a `Sequence[Skill]` |
| `cosmo_ai.mcp` | `McpStdioServer`, `McpConfigError`, `McpExtraNotInstalled`, `McpInput` — the `mcp=` argument takes a `.mcp.json` path or a list of servers |
| `cosmo_ai.hooks` | the four seam decorators (`@hooks.session_start`, `@hooks.pre_tool_use(matcher=…)`, …), `Hook`, the context/result types, the `ToolOk`/`ToolError`/`ToolDenied` outcomes, and the server hooks (`SilenceTimeout`, `Say`, `EndCall`) |
| `cosmo_ai.errors` | `CosmoRealtimeError` (the base every SDK error extends), `SessionStartError`, `VersionMismatchError`, `MintTokenError`, `DialError`, `NotConnectedError`, `ExtraNotInstalledError` |

The wire models (internal, `_internal/protocol.py`) are hand-written Pydantic
mirrors of the published OpenAPI spec; `tests/test_spec_pin.py` fails on any
drift between the two. The cross-language behavioral contract — the same trace
files every Cosmo SDK replays — runs via
`tests/contract/test_external_traces.py`.

---

## Skills

Decompose a long system prompt into just-in-time **skills**. Each skill is a
`SKILL.md` (name + description + body, the Agent Skills standard); only the
names/descriptions ride in the prompt, and the body loads on demand via a
single `load_skill` tool. A loaded body stays in context for the rest of the
call and counts toward the prompt every turn, so keep bodies tight (split long
walkthroughs into small skills).

Attach skills with the `skills` argument — a directory, or a list mixing
directories and inline `Skill` objects (a path element expands in place, so
built-in skills can layer with a user folder:
`skills=[*BUILTINS, "./user-skills"]`; duplicate names raise):

```python
from cosmo_ai import CosmoRealtime

client = CosmoRealtime(api_key="cosmo_...")
agent = client.agent(instructions="You are Alex.", skills="./skills")
```

```python
from cosmo_ai.skills import Skill

agent = client.agent(
    instructions="You are Alex.",
    skills=[Skill(name="activate-card", description="...", body="...")],
)
```

Skill files live in `./skills/<name>/SKILL.md`:

```markdown
---
name: activate-card
description: Walk the customer through activating their card.
---
Acknowledge they want to activate. Ask web or app, then one step at a time.
```

Directory semantics: a directory that itself contains a `SKILL.md` is that one
skill; otherwise each `<child>/SKILL.md` is a skill. A directory yielding no
skills logs a warning and attaches none (an empty per-user skills folder is a
valid state); a missing path or malformed SKILL.md raises `SkillParseError`
when the agent is built, not mid-call. Unknown frontmatter keys (`tier`,
`allowed-tools`, `license`, …) are ignored, so files authored for other
harnesses stay valid.

---

## Development

```bash
pip install -e ".[dev]"
ruff check src/
mypy src/
pytest
```
