Build a Python voice CLI

Install cosmo-ai-sdk, connect with CosmoRealtime → agent.start(), enable the OS mic and speaker, stream typed events, exit cleanly on Ctrl-C.

This guide walks through building a terminal voice client with cosmo-ai-sdk. By the end you will have a script that starts a Cosmo session, captures the default OS microphone, plays the agent's voice on the default speaker, prints live transcripts, and exits gracefully.

Base code: sdks/cosmo-realtime/python/examples/voice_cli/.

Prerequisites

Python 3.11+
A Cosmo API key (cosmo_…)

1. Create a project

mkdir my-voice-cli && cd my-voice-cli
python -m venv .venv && source .venv/bin/activate

2. Install the package

Room connectivity (livekit-rtc) and OS audio I/O (sounddevice, used by set_microphone_enabled / set_speaker_enabled — livekit-rtc Python has no native capture or playback) are both included in the base install. On Linux, OS audio also needs the system PortAudio library (apt install libportaudio2).

pip install cosmo-ai-sdk

For development from the monorepo:

pip install -e sdks/cosmo-realtime/python

On Linux, sounddevice needs the PortAudio native library (apt install libportaudio2). The macOS and Windows wheels bundle it.

3. Write the client

Three objects, three concerns: CosmoRealtime is the connection (credential + endpoint), client.agent(...) is the reusable persona, and agent.start() opens one live run. The session is an async iterator of typed events.

# voice_cli.py
from __future__ import annotations

import argparse
import asyncio
import os
import sys

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


async def print_events(session: RealtimeSession) -> None:
    async for event in session:
        if isinstance(event, RealtimeReady):
            print(f"[ready] session_id={session.session_id}", flush=True)
            print(
                "Speak into your microphone — the agent replies out loud. "
                "Press Enter (blank line) to end.",
                flush=True,
            )
        elif isinstance(event, RealtimeTranscriptDelta):
            marker = "»" if event.is_final else "…"
            print(f"  [{event.role.value}] {event.text}{marker}", flush=True)
        elif isinstance(event, RealtimeToolCall):
            print(f"  [tool] {event.name} (id={event.tool_call_id})", flush=True)
        elif isinstance(event, RealtimeError):
            print(
                f"  [error] {event.code.value}: {event.message}",
                file=sys.stderr,
                flush=True,
            )
        elif isinstance(event, RealtimeSessionEnded):
            print(f"  [ended] {event.reason}", flush=True)


async def run(api_key: str, voice: str | None, model: str | None) -> None:
    async with CosmoRealtime(api_key=api_key) as client:
        print("Connecting…", flush=True)
        agent = client.agent(voice=voice, model=model)
        async with agent.start() as session:
            print(f"Joined room: {session.response.room_name}", flush=True)

            printer = asyncio.create_task(print_events(session))

            print("Enabling microphone…", flush=True)
            await session.set_microphone_enabled(True)

            print("Enabling speaker…", flush=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

            print("Ending session…", flush=True)
            await session.end()
            await printer

Notes:

There is no project ID to pass — the API key scopes the session server-side.
agent.start() is both awaitable (session = await agent.start(), you call session.end() yourself) and an async context manager (the canonical form — the session ends on exit).
Events consume from one place: the async for loop. RealtimeSessionEnded is always the final item, after which iteration finishes — so await printer returns once the session is over.
Unknown frame types surface as UnknownEvent and never end the stream.

4. Add the CLI entrypoint

# voice_cli.py (continued)

def main() -> None:
    parser = argparse.ArgumentParser(description="Cosmo voice CLI")
    parser.add_argument(
        "--api-key",
        default=os.environ.get("COSMO_API_KEY", ""),
        help="Bearer API key (or set COSMO_API_KEY)",
    )
    parser.add_argument("--voice", default=None, help="Provider voice id (optional)")
    parser.add_argument(
        "--model", default=None, help="Model id (optional; see GET /models)"
    )
    args = parser.parse_args()

    if not args.api_key:
        parser.error("--api-key is required (or set COSMO_API_KEY)")

    asyncio.run(run(api_key=args.api_key, voice=args.voice, model=args.model))


if __name__ == "__main__":
    main()

The SDK targets https://app.askcosmo.ai by default; set the COSMO_BASE_URL environment variable to point at another backend for local development (http:// is allowed only for localhost).

5. Run it

export COSMO_API_KEY=cosmo_...
python voice_cli.py

You will see:

Connecting…
Joined room: session-abc123
Enabling microphone…
Enabling speaker…
[ready] session_id=sess_…
Speak into your microphone — the agent replies out loud. Press Enter (blank line) to end.
  [USER] Hello Cosmo»
  [ASSISTANT] Hi there! How can I help you today?»
Ending session…
  [ended] client ended

Press Enter to end the session. Ctrl-C also works — the KeyboardInterrupt/EOFError path in run() falls through to the same session.end() teardown, and the async with blocks guarantee cleanup even on an unexpected exception.

6. Give the agent instructions and tools

The persona lives on the agent. Declare client-executed tools with the @tool decorator (a Pydantic model drives the schema and validation) and opt in to server-executed tools with their typed specs:

from typing import Any

from pydantic import BaseModel, Field

from cosmo_ai import CosmoRealtime, WebSearchTool, tool


class WeatherInput(BaseModel):
    city: str = Field(description="City name")


@tool
async def get_weather(input: WeatherInput) -> dict[str, Any]:
    """Current weather for a city."""
    return {"temp_c": 21.5}


agent = client.agent(
    instructions="You are a terse assistant.",
    voice="Puck",
    tools=[get_weather, WebSearchTool()],
)

tools=None (the default) runs the session with no tools. Server tools the backend refuses are echoed on RealtimeReady.rejected_tools and the session starts without them. See Server-side tools and Tools.

Derive a variant without re-specifying the rest with agent.with_(voice="Aoede").

7. Send text turns

When the microphone is not needed (testing, accessibility), use send_text on the session instead of audio:

await session.send_text("What's the weather in Paris?")

Send a text turn the same way you would any other:

await session.send_text("Summarise this document")

8. Observe the lifecycle

agent.start(on_state_change=...) takes a callback that fires on every transport-lifecycle transition (idle → connecting → connected ↔ reconnecting → disconnected):

from cosmo_ai import SessionState


def on_state(state: SessionState) -> None:
    print(f"[state] {state.kind.value}", flush=True)


async with agent.start(on_state_change=on_state) as session:
    ...

On DISCONNECTED, state.disconnect_reason and state.detail say why. See Reconnects.

9. Packaging as a CLI tool

Add a pyproject.toml to distribute the script:

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "my-voice-cli"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["cosmo-ai-sdk"]

[project.scripts]
my-voice = "voice_cli:main"

Install and run:

pip install -e .
my-voice --api-key cosmo_...

Next steps

Handle tool calls — match tool events in the stream
Server-side tools — opt in to Cosmo-executed tools
Tools — the full tool model across SDKs
Debugging — correlate logs with session IDs

Build a Python voice CLI

On this page