RealtimeClient (Python)

The Python SDK is three objects, one per concern:

• RealtimeClient — how to reach Cosmo: credential, endpoint, HTTP transport.
• RealtimeAgent — 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 RealtimeClient, TranscriptDeltaEvent

client = RealtimeClient(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 TranscriptDeltaEvent():
                print(event.role, event.text, event.is_final)

RealtimeClient

RealtimeClient(
    *,
    api_key: str | None = None,
    token: str | TokenSource | None = None,
    transport: Literal["webrtc", "websocket", "livekit"] | None = None,
    http_client: httpx.AsyncClient | None = None,
)

Pass at most one credential — passing both raises CredentialsError (CONFLICTING_CREDENTIALS). Passing neither is the recommended local form: the SDK resolves a credential itself, reading COSMO_API_KEY first and falling back to the ~/.cosmo/credentials file cosmo login writes. See API keys.

The websocket transport carries audio as PCM over the same socket as the session protocol. What that costs is what a media room provides: no reconnection and no camera or screen input. The microphone is echo-cancelled; noise suppression and gain control are not applied. It runs only against an OSS cosmo-server on your own machine: that server is for local development, and managed Cosmo does not serve the socket route.

The API is reached at https://platform.askcosmo.ai by default. Set the COSMO_BASE_URL environment variable to point elsewhere (http:// is allowed only for localhost). A credential resolved from the credentials file carries its own base_url, which outranks COSMO_BASE_URL; when the two name different origins the constructor raises CredentialsError with code BASE_URL_MISMATCH rather than sending a stored key somewhere it was not issued for. Supports async with; aclose() releases the owned HTTP client.

agent(...)

client.agent(
    *,
    instructions: str | None = None,
    model: RealtimeModel | None = None,
    voice: str | VoiceConfig | None = None,
    tools: Sequence[AgentTool] | 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,
    plugins: Sequence[Plugin] | None = None,
) -> RealtimeAgent

Build an inline RealtimeAgent. 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). See Tools, Skills, Hooks, and MCP for the composable inputs.

model is a model id or provider alias as a plain string, or one provider block discriminated on provider. A block pins the concrete model with model_id and carries that provider's knobs:

Leave model_id unset to run the provider's default model; a model_id belonging to another provider is rejected at session start. On GeminiModel, model_id="gemini-3.8-live" runs Gemini 3.8 Live, which takes no thinking_level: setting one fails session start with thinking_level_unsupported. The endpointing knobs bind how long the server waits before deciding the user's turn is over — see Turn-taking. Pairing an OpenAI knob with the detector that isn't selected is rejected at session start.

Plugins bundle instructions, skills, tools, and hooks for inline agents. Pass plugins at construction; the built agent exposes the expanded contributions. See Plugins for the bundle fields, merge order, and PluginError codes.

catalog_agent(...)

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

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 aren't 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, *, ttl_seconds: int | None = None
) -> MintedToken  # .jwt, .expires_at, .token_id

Mint a short-lived end-user token (requires an api_key client). Hand the jwt to the end user's device, which constructs RealtimeClient(token=jwt). Idempotent per (workspace, external_user_id). ttl_seconds (60–86400) shortens the 24-hour default lifetime; token_id is the revocation handle (DELETE /api/v1/external/auth/token/{token_id}) — keep it server-side. 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:start, and realtime_voice_available whether this deployment has the default voice stack configured. workspace is None for a minted token — an end user isn't told the workspace it belongs to. Raises VerifyError only when the server rejects the credential. See API keys.

get_session_usage(...)

await client.get_session_usage(session_id: str) -> SessionUsage

Fetch any session's usage summary by id. session.usage() is the same read from a live session object; this is the form for a process that no longer holds one — a later run, a nightly billing job, a recovery after a crash. Raises UsageError. See GET /sessions/{session_id}/usage.

RealtimeAgent

A frozen dataclass holding the persona fields (instructions, model, voice, tools, interruption_sensitivity, greeting, audio, mcp, skills, hooks, plus catalog-only name / inputs). Build one with client.agent(...) or client.catalog_agent(...). An agent is immutable and opens any number of sessions; to vary a persona field, build another one.

start(...)

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

Start one live session. Resolves once the session is ready — the server's handshake has landed, so every session method works immediately. 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]).

store_audio, store_transcript, and store_video opt out of one artifact class each and take precedence over store_recording. All four narrow only — a run may store less than the account's consents allow, never more. See Recording and privacy.

The return value is a SessionHandle (a root export, for annotating code that wraps start()): both an async context manager (ends the session on exit — the canonical form) and awaitable (you own the lifecycle; call session.end() yourself), either form yielding the ready RealtimeSession. The SDK creates it; there is no reason to construct one yourself.

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

Every failed start raises SessionStartError, whose code is a closed SessionStartErrorCode naming how far the attempt got. Once the transport has joined, three more exits are typed, and each tears the session down before it raises — you never hold a session for a run that failed to become ready:

A pre-ready error frame never raises on its own: it is stashed, and supplies the server code and message of the handshake failure that the close then raises.

prepare_session(...)

agent.prepare_session(
    *,
    resume_session_id: UUID | str | None = None,
    store_recording: bool | None = None,
    store_audio: bool | None = None,
    store_transcript: bool | None = None,
    store_video: bool | None = None,
    on_state_change: OnStateChange | None = None,
)

Prepare one session ahead of its start, so it starts faster. The SDK reserves a room in the background immediately, and the returned PreparedSession joins it while the session request is still in flight when you start it — instead of waiting for a room to be allocated after the request returns. The arguments are the same per-run options start() takes; they are fixed here, and the start takes none.

Prepare as early as the app knows a session is coming — while the rest of its setup runs — and start when the user is ready:

prepared = agent.prepare_session()
...  # the rest of the app's setup
async with prepared.start() as session:
    ...

Purely an accelerator: a reservation that failed, lapsed, or is declined by the server leaves the start on the ordinary path, with the same result as start(). Needs a running event loop (raises RuntimeError otherwise) and the webrtc transport (raises ValueError on the websocket lane, which has no rooms to prepare).

PreparedSession

Returned by agent.prepare_session(): one session prepared ahead of its start. The reservation is refreshed in the background until the handle is started or closed, so one held for hours stays warm.

start()

Start the prepared session. Returns the same SessionHandle as agent.start() — an async context manager or an awaitable, either form yielding the ready RealtimeSession — resolving at ready and raising on the same terms. Single-use: a second call raises RuntimeError; prepare another session for another start.

close()

await prepared.close()

Drop the reservation and stop refreshing it, for a handle that will never be started. A no-op once started.

RealtimeSession

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

Properties

Methods

await session.send_text(content: str, *, transcript: bool = True)

Send a text turn — the agent answers it. For a session that never speaks, configure the agent with audio=AudioConfig(output=False). The sent text lands in session.transcript as its own closed user turn — an in-progress speech transcription is untouched. transcript=False keeps it out.

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 content is validated client-side before anything is sent: 1–4096 characters, enforced by the wire model (a pydantic ValidationError), never truncated.

await session.append_thinking(content: str, *, delegation_id: str | None = None)
await session.append_commentary(content: str, *, delegation_id: str | None = None)
await session.append_instructions(content: str, *, delegation_id: str | None = None)

Hand text to the voice model on a GPT Live session that delegates work (delegation="client" or "cosmo"). append_thinking is background the model keeps to itself and draws on when relevant; append_commentary is something it says now, in its own words; append_instructions changes how it behaves from here on. delegation_id names the DelegationCreatedEvent the text answers; None steers the session as a whole. Each append is one short piece — send several as work progresses rather than one long one at the end. Content is 1–4096 characters, enforced by the wire model.

await session.set_muted(muted: bool)

Toggle the server-side mic gate.

await session.ping()

Heartbeat; the server replies with a PongEvent event.

await session.send_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.usage() -> SessionUsage

Fetch this session's usage summary — duration, talk time, and token counts in provider-reported units. An authenticated REST read, callable while the session is live and after it ends. usage_status on the result reports whether the detailed summary is present: UsageStatus.PENDING while it may still land (poll), RECORDED once the numbers are final, UNAVAILABLE when none was written and none will be. tokens is None when the provider does not report token usage. Raises UsageError. See GET /sessions/{session_id}/usage.

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, start_audio_stream — are covered on the Audio reference. Screen-share methods (start_screen_share, push_screen_share_frame, stop_screen_share) and the video-stream methods (add_video_stream, remove_video_stream) are covered in Screen share and Video.

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. ToolInvocationEvent is the observability event for client-executed tools. See Realtime events for the protocol-level view.

Errors

All exceptions derive from RealtimeError. Most are importable from cosmo_ai; the feature-scoped ones live beside the features that raise them — ToolDefinitionError and ToolInputValidationError from cosmo_ai.tools, HookError from cosmo_ai.hooks, SkillError from cosmo_ai.skills, and McpError from cosmo_ai.mcp.

See Errors for recovery patterns and Python types for the credentials and token-source type detail.