Python Quickstart

Prerequisites

Python 3.10+
A workspace API key with the realtime:use scope — create one in the dashboard (API keys)

Install

pip install cosmo-ai-sdk

The WebRTC transport and OS microphone/speaker I/O are included. On Linux, OS audio also needs the system PortAudio library (apt install libportaudio2).

Set your API key

export COSMO_API_KEY=cosmo_...

The SDK targets https://app.askcosmo.ai by default; set COSMO_BASE_URL to point at another backend. The API key scopes your workspace server-side — there is no project ID to pass.

Write the script

CosmoRealtime is the client; client.agent(...) builds a reusable agent; agent.start() opens one live session as an async context manager. The session is an async iterator of typed events.

import asyncio
import os
import sys

from cosmo_ai import (
    CosmoRealtime,
    RealtimeError,
    RealtimeReady,
    RealtimeSession,
    RealtimeSessionEnded,
    RealtimeTranscriptDelta,
)


async def print_events(session: RealtimeSession) -> None:
    async for event in session:
        if isinstance(event, RealtimeReady):
            print("Ready — speak into your microphone. Press Enter to end.")
        elif isinstance(event, RealtimeTranscriptDelta):
            marker = "»" if event.is_final else "…"
            print(f"[{event.role.value}] {event.text}{marker}", flush=True)
        elif isinstance(event, RealtimeError):
            print(f"[error] {event.code.value}: {event.message}", file=sys.stderr)
        elif isinstance(event, RealtimeSessionEnded):
            print(f"Session ended: {event.reason}")


async def main() -> None:
    async with CosmoRealtime(api_key=os.environ["COSMO_API_KEY"]) as client:
        agent = client.agent(
            instructions="You are a terse voice assistant.",
            greeting="Hi — how can I help?",
        )
        async with agent.start() as session:
            print(f"Connected — session_id={session.session_id}")

            printer = asyncio.create_task(print_events(session))

            await session.set_microphone_enabled(True)
            await session.set_speaker_enabled(True)

            loop = asyncio.get_event_loop()
            try:
                await loop.run_in_executor(None, sys.stdin.readline)
            except (KeyboardInterrupt, EOFError):
                pass

            await session.end()
            await printer


if __name__ == "__main__":
    asyncio.run(main())

RealtimeSessionEnded is the stream's terminal event, so the printer task finishes on its own once session.end() runs. Leaving the async with block also ends the session.

Run it

python quickstart.py

What you should see

Connected — session_id=rs_...
Ready — speak into your microphone. Press Enter to end.
[ASSISTANT] Hi — how can I help?»
[USER] what's the tallest mountain…
[USER] what's the tallest mountain in the world»
[ASSISTANT] Mount Everest, at 8,849 meters.»
Session ended: client_ended

No microphone handy? Drive the session over the text channel instead: await session.send_text("Hello!") sends a text turn. The hello_realtime.py example in the SDK repo (sdks/cosmo-realtime/python/examples/) is a complete text-only version, and examples/voice_cli/ is the full voice CLI this page is based on.

Session methods

Method	What it does
`send_text(content)`	Send a text turn the agent answers.
`send_context(content)`	Give the agent context without asking it anything — no turn, no speech.
`set_microphone_enabled(enabled)`	Publish or unpublish the OS microphone.
`set_speaker_enabled(enabled)`	Play or stop the agent's audio on the OS speaker.
`end()`	End the session gracefully (also runs on `async with` exit).

Next steps

Clients, agents, and sessions — the three-tier model in depth
Events — every typed event the session yields
Tools — @tool-decorated Python functions the agent can call
Hooks — intercept session start, tool use, and speech timeouts
Telephony — put the same agent on a phone call

Python Quickstart

On this page