cosmo-ai-sdk 0.1.0

Cosmo Realtime SDK — typed event-stream client for voice + multimodal sessions

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 Developer Guide — getting started, the credential model, and the expected session lifecycle.

---

Install

pip install "cosmo-ai-sdk[livekit]"

Beta. The SDK is pre-1.0: minor releases may include breaking API changes, called out in the 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.1.0) if you need stability.

Requires Python 3.10+. The [livekit] extra pulls in livekit>=0.18 (required for joining sessions; the core package alone gives you the typed protocol models). The [audio] extra adds OS audio I/O — microphone capture via set_microphone_enabled and speaker playback via set_speaker_enabled (sounddevice, plus livekit — it stands alone): pip install "cosmo-ai-sdk[audio]".

---

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

import asyncio
from cosmo_ai import (
    CosmoRealtime,
    RealtimeReady,
    RealtimeSessionEnded,
    RealtimeTranscriptDelta,
)

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!", audio_response=False)

        async for event in session:
            match event:
                case RealtimeReady():
                    print(f"ready — session {event.session_id}")
                case RealtimeTranscriptDelta():
                    print(f"[{event.role.value}] {event.text}")
                case RealtimeSessionEnded():
                    print(f"ended: {event.reason}")

asyncio.run(main())

See examples/hello_realtime.py for the runnable text-only demo and examples/voice_cli/ for a live two-way voice call (microphone in, agent audio out).

The client talks to https://app.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:

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

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.

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 — all Pydantic models discriminated by type.

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.

Public API

CosmoRealtime

Member	Description
CosmoRealtime(api_key=... | token=..., http_client=None)	Construct with exactly one credential (see Credentials). The base URL defaults to https://app.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, temperature=None, max_output_tokens=None, thinking_level=None, voice=None, turn_endpoint_delay_seconds=None, tools=None, interruption_sensitivity=None, greeting=None, noise_cancellation_enabled=None, mcp=None, skills=None, hooks=None, screen_interaction=None)	Build a reusable inline Agent — the persona (see Agent), including its opening greeting and its inbound-audio handling (noise_cancellation_enabled). tools is a list of ClientTool / typed server opt-ins (WebSearchTool, ExamineImageTool, DetectTool, PointTool); mcp / skills / hooks / screen_interaction attach the concepts described in their sections below. Fields left None fall back to the server default
catalog_agent(name, *, inputs=None, voice=None, tools=None, mcp=None, hooks=None, screen_interaction=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 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
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 noise_cancellation_enabled — ride unchanged from the agent (derive a with_ variant 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, ExtraNotInstalledError without the [livekit] extra
with_(*, …)	Derive a new agent with selected persona fields overridden (same keyword set as agent(...)); fields left None keep this agent's values

RealtimeSession

Member	Description
async for event in session	The typed event stream (see above)
send_text(content, *, audio_response=True)	Send text instead of audio
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. Needs the [audio] extra (livekit-rtc Python has no native mic capture; the SDK supplies it via sounddevice); raises ExtraNotInstalledError without it. See examples/voice_cli
set_speaker_enabled(enabled)	Play the agent's voice on the default OS output device (or stop it). Needs the [audio] extra (livekit-rtc Python has no native playback; the SDK supplies it via sounddevice); raises ExtraNotInstalledError without it. See examples/voice_cli
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 AgentAudioFrames (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)
start_screen_share() / push_screen_share_frame(frame) / stop_screen_share()	Screen-share publish
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.

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 patterns, formats, 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):

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. A spec without a handler is still declared to the agent but cannot be executed — its invocation surfaces only as a RealtimeToolInvocation observability event on the stream.

Typed opt-in classes enable built-in server-executed tools — WebSearchTool (web search), ExamineImageTool (full-resolution frame examination), DetectTool / PointTool (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.

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.

async with agent.start() as session:        # session only — no participants yet
    await session.dial("[phone]")        # 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 examples/outbound_call.py.

---

Screen interaction

Give the agent grounded control of your app's screen. The model calls a server tool; the server captures the screen (via your conforming type), grounds the natural-language target to one element with a vision model, then asks your client to act on it or point at it. You implement only the platform pieces — the ScreenInteraction protocol (capture, activate, highlight) — and the SDK owns the transport plumbing (RPC registration, streaming each capture out of band from the JSON control channel). Pass one as screen_interaction= to opt in; omit it and the tools are never offered.

from cosmo_ai.screen_interaction import (
    ScreenAction,
    ScreenCapture,
    ScreenElement,
)

class AppScreen:  # conforms to the ScreenInteraction protocol
    async def capture(self) -> ScreenCapture: ...
    async def activate(
        self, element: ScreenElement, capture: ScreenCapture, action: ScreenAction
    ) -> bool: ...
    async def highlight(
        self,
        element: ScreenElement,
        capture: ScreenCapture,
        *,
        label: str,
        placement: str,
        interaction: str,
    ) -> bool: ...

agent = client.agent(instructions="You are Alex.", screen_interaction=AppScreen())

activate/highlight return True on success and False when safely declined (so the model can retry); raise only on unexpected failure.

Platform-neutral: macOS clicks a mouse, iOS taps, a web client clicks the DOM — the wire, the grounder, and the contract don't change; only your conforming type's methods differ per platform. The full protocol + value types live in cosmo_ai.screen_interaction; the wire mechanics are pinned by the shared screen-interaction vectors.

---

MCP servers (local stdio)

Expose any local MCP 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>.

from cosmo_ai import CosmoRealtime

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

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:

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 examples/hooks_agent.py 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), 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.screen_interaction	ScreenInteraction (the platform-conformer protocol) + its value types (ScreenCapture, ScreenElement, ScreenAction, ScreenButton)
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 sdks/cosmo-realtime/external-openapi.json; tests/test_spec_pin.py fails on any drift between the two. The cross-language behavioral contract lives in sdks/cosmo-realtime/contract/ and 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):

from cosmo_ai import CosmoRealtime

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

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:

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

pip install -e ".[dev,livekit]"
ruff check src/
mypy src/
pytest