CosmoRealtime (Python)

The Python SDK is three objects, one per concern:

• CosmoRealtime — how to reach Cosmo: credential, endpoint, HTTP transport.
• Agent — the persona/configuration of the model on the other end: instructions, model, voice, tools, turn-taking. Immutable, reusable across runs.
• RealtimeSession — one live run: a typed async event stream plus every mid-call send.

from cosmo_ai import CosmoRealtime, RealtimeTranscriptDelta

client = CosmoRealtime(api_key="cosmo_...")
agent = client.agent(instructions="You are terse.", voice="Puck")
async with agent.start() as session:
    async for event in session:
        match event:
            case RealtimeTranscriptDelta():
                print(event.role, event.text, event.is_final)

CosmoRealtime

CosmoRealtime(
    *,
    api_key: str | None = None,
    token: str | None = None,
    http_client: httpx.AsyncClient | None = None,
)

Construct with exactly one credential — passing both or neither raises ValueError.

The API is reached at https://app.askcosmo.ai by default. Set the COSMO_BASE_URL environment variable to point elsewhere (http:// is allowed only for localhost). Supports async with; aclose() releases the owned HTTP client.

agent(...)

client.agent(
    *,
    instructions: str | None = None,
    model: str | None = None,
    model_options: RealtimeModelOptions | None = None,
    voice: str | VoiceConfig | None = None,
    tools: Sequence[ClientTool | ServerTool] | None = None,
    interruption_sensitivity: InterruptionSensitivity | None = None,
    greeting: str | None = None,
    audio: AudioConfig | None = None,
    mcp: McpInput | None = None,
    skills: SkillsInput | None = None,
    hooks: Sequence[Hook | ServerHook] | None = None,
) -> Agent

Build an inline Agent. Fields left None fall back to the protocol's server-side defaults. voice takes the voice id as a plain string, or a VoiceConfig(name=..., speaking_style=...); audio is an AudioConfig (output, noise_cancellation, and ambience — an AmbienceConfig whose presence enables the bed). See Tools, Skills, Hooks, and MCP for the composable inputs.

catalog_agent(...)

client.catalog_agent(
    name: str,
    *,
    inputs: Mapping[str, str] | None = None,
    voice: str | VoiceConfig | None = None,
    tools: Sequence[ClientTool | ServerTool] | None = None,
    mcp: McpInput | None = None,
    hooks: Sequence[Hook] | None = None,
) -> Agent

Run a workspace catalog agent by its machine handle — the server resolves name at session start and runs the stored config verbatim. Only per-run ride-alongs are accepted: inputs (values for the agent's declared input fields), voice (cosmetic override), tools / mcp (client-executed declarations), and hooks (local only — server hooks are not accepted here). Stored-config fields like instructions have no parameter, so the illegal combination is a type error.

mint_token(...)

await client.mint_token(external_user_id: str) -> MintedToken  # .jwt, .expires_at

Mint a short-lived end-user token (requires an api_key client). Hand the jwt to the end user's device, which constructs CosmoRealtime(token=jwt). Idempotent per (workspace, external_user_id). Raises MintTokenError. See End-user credentials.

verify()

await client.verify() -> CredentialInfo
# .credential, .workspace (name/slug, None for a token), .scopes,
# .can_start_sessions, .realtime_voice_available, .external_user_id

Check the credential without starting a session — free, and works with either credential. Returning at all means the credential authenticated; can_start_sessions reports whether it carries realtime:use, and realtime_voice_available whether this deployment has the default voice stack configured. workspace is None for a minted token — an end user is not told the workspace it belongs to. Raises VerifyError only when the server rejects the credential. See API keys.

Agent

A frozen dataclass holding the persona fields (instructions, model, model_options, voice, tools, interruption_sensitivity, greeting, audio, mcp, skills, hooks, plus catalog-only name / inputs). Build one with client.agent(...) or client.catalog_agent(...).

with_(...)

agent.with_(*, instructions=..., voice=..., ...) -> Agent

Derive a new agent with selected persona fields overridden; fields left None keep this agent's value. Accepts the same keyword arguments as client.agent(...).

start(...)

agent.start(
    *,
    resume_session_id: UUID | str | None = None,
    store_recording: bool | None = None,
    on_state_change: OnStateChange | None = None,
)

Start one live session. The arguments are the per-run, transport-level concerns: resume_session_id continues a prior session, store_recording=False opts this run out of server-side recording artifacts, on_state_change observes the lifecycle (a Callable[[SessionState], None]).

The return value is both an async context manager (ends the session on exit — the canonical form) and awaitable (you own the lifecycle; call session.end() yourself):

async with agent.start() as session: ...
session = await agent.start()

Raises VersionMismatchError when the server refuses the protocol version, SessionStartError for any other rejection, and ExtraNotInstalledError when the livekit extra is missing.

RealtimeSession

Async-iterate it for events; call its send methods to talk back. Unrecognized frames surface as UnknownEvent and never end the stream; RealtimeSessionEnded is always the final item, after which iteration finishes.

Properties

Methods

await session.send_text(content: str)

Send a text turn — the agent answers it. For a session that never speaks, configure the agent with audio.output = false.

await session.send_context(content: str)

Give the agent context without asking it anything: no turn, no speech, no transcript entry — it simply knows this the next time it answers. For live application state (scroll position, selection, current record). The server rejects a note longer than 4096 characters rather than truncating it.

await session.set_muted(muted: bool)

Toggle the server-side mic gate.

await session.ping()

Heartbeat; the server replies with a RealtimePong event.

await session.activity_end()

Signal end-of-turn for manual-VAD turn-taking. See Turn-taking.

await session.send_image(*, data: str, mime_type: str = "image/jpeg",
                         stream_id: str = "video.input.default")

Send one base64-encoded image frame. See Image input.

await session.dial(phone_number: str, *, caller_number: str | None = None) -> DialResult

Place an outbound phone call into this live session's room. phone_number must be E.164 (+ then 8–15 digits, validated locally before the request); caller_number is an optional E.164 caller-ID from the workspace pool. Returns once the dial is queued (DialResult.dial_id); the call rings asynchronously. Raises DialError. See Telephony.

await session.end()    # graceful: tells the server to tear down; idempotent
await session.close()  # abrupt local teardown without telling the server; idempotent

Audio I/O methods — set_microphone_enabled, set_speaker_enabled, set_agent_playback_volume, agent_audio, audio_levels, publish_audio_source — are covered on the Audio reference. Screen-share methods (start_screen_share, push_screen_share_frame, stop_screen_share) are covered in Screen share.

Events

Everything async for event in session can yield (the RealtimeSessionEvent union). Match on the class:

Server tools follow the three-event lifecycle tool-call → tool-dispatch-started → tool-result, correlated by tool_call_id. RealtimeToolInvocation is the observability event for client-executed tools. See Realtime events for the protocol-level view.

Errors

All exceptions derive from CosmoRealtimeError and are importable from cosmo_ai (the two tool errors from cosmo_ai.tools / cosmo_ai.errors).

See Errors for recovery patterns.