Build a Python voice CLI

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

Build a terminal voice client with cosmo-ai-sdk. By the end you 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: examples/python/voice_cli/ in the examples repo.

Prerequisites

Python 3.10+
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 (the livekit package, imported as livekit.rtc) and OS audio I/O are both included in the base install. set_microphone_enabled captures through WebRTC's audio device module; set_speaker_enabled plays through sounddevice, which on Linux needs the system PortAudio library (apt install libportaudio2).

pip install cosmo-ai-sdk

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

3. Write the client

Three objects, three concerns: RealtimeClient 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 (
    RealtimeClient,
    ErrorEvent,
    ReadyEvent,
    RealtimeSession,
    SessionEndedEvent,
    ToolCallEvent,
    TranscriptDeltaEvent,
)


async def print_events(session: RealtimeSession) -> None:
    async for event in session:
        if isinstance(event, ReadyEvent):
            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, TranscriptDeltaEvent):
            marker = "»" if event.is_final else "…"
            print(f"  [{event.role.value}] {event.text}{marker}", flush=True)
        elif isinstance(event, ToolCallEvent):
            print(f"  [tool] {event.name} (id={event.tool_call_id})", flush=True)
        elif isinstance(event, ErrorEvent):
            print(
                f"  [error] {event.code.value}: {event.message}",
                file=sys.stderr,
                flush=True,
            )
        elif isinstance(event, SessionEndedEvent):
            print(f"  [ended] {event.reason}", flush=True)

print_events drains the session's one event stream. run owns the session itself — open it, start the printer, publish audio, wait for a blank line, end:

# voice_cli.py (continued)

async def run(api_key: str, voice: str | None, model: str | None) -> None:
    async with RealtimeClient(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"Session started: {session.session_id}", 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:

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. SessionEndedEvent 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)"
    )
    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://platform.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 see:

Connecting…
Joined room: cosmo-9f3c1ad84be7205c6d18e4b2
Enabling microphone…
Enabling speaker…
[ready] session_id=3f2b8c1e-9a4d-4e7f-8b21-06d5c9a1f4e2
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 RealtimeClient, web_search_tool, 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, web_search_tool()],
)

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

An agent is immutable; to run a different voice or prompt, build a second one with client.agent(...).

7. Send text turns

When the microphone isn't 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. Package it 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 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