Agent recipes

The Python examples directory holds a set of single-file scripts, each demonstrating one agent capability in the fewest possible lines. They all share the same shape — build a client, configure an agent, iterate the session's event stream — so once you've run one, the others read as diffs.

Full source: examples/python in the examples repo. The Swift siblings of these recipes live in the HelloRealtime package as the SkillsExample, HooksExample, and MCPExample targets.

Prerequisites

• Python 3.10+
• pip install cosmo-ai-sdk (the MCP recipe needs pip install "cosmo-ai-sdk[mcp]" and Node for npx)
• A workspace API key with the realtime:start scope (API keys), or a prior cosmo login
• For the outbound-call recipe only: the key also needs realtime:dial, and telephony enabled on your workspace (Telephony)

Each script resolves credentials on its own — RealtimeClient() with no arguments finds COSMO_API_KEY or the cosmo login credentials file — so "run it" is one command per recipe.

Run a text-only session with a typed client tool

hello_realtime.py needs no microphone or speaker: it drives the agent over the text channel and prints every event. Its one tool shows the @tool decorator, where a pydantic input model supplies both the model-facing schema and the validated arguments your handler receives:

from pydantic import BaseModel, Field
from cosmo_ai import RealtimeClient, tool

class TimeInput(BaseModel):
    label: str = Field(default="now", description="Label echoed back with the time")

@tool
async def get_current_time(input: TimeInput) -> dict[str, Any]:
    """Current UTC time, for questions about the time or date."""
    return {"label": input.label, "utc": datetime.now(timezone.utc).isoformat()}

async with RealtimeClient() as client:
    agent = client.agent(instructions="You are a terse assistant.", tools=[get_current_time])
    async with agent.start() as session:
        await session.send_text("Hello from the Python SDK! What time is it?")
        async for event in session:
            ...

Run python hello_realtime.py. It sends one text turn, prints the transcript and the get_current_time round-trip, then ends itself. The output has this shape — session and tool-call ids and the agent's wording vary run to run:

Connected — session_id=b4a4a0f2-6d1e-4b0a-8f3c-2e9d51c7a8b0
Sending text message…
[ready] session_id=b4a4a0f2-6d1e-4b0a-8f3c-2e9d51c7a8b0
[transcript:user] Hello from the Python SDK! What time is it?»
[tool_call] get_current_time (id=call_a1b2c3)
[tool_result:ok] {'label': 'now', 'utc': '2026-08-05T21:14:03.512306+00:00'}
[transcript:assistant] It's 21:14 UTC.»
[session-ended] client ended
Session ended.

The agent still speaks into the room even in this mode — nothing here plays it. Configure audio=AudioConfig(output=False) for a genuinely silent session.

Attach skills that load just in time

skills_agent.py points the agent at a directory of SKILL.md files — the sibling skills/ directory holds two, activate-card and report-lost-card. The agent sees each skill's name and description up front and pulls in the full body only when the conversation needs it:

agent = client.agent(
    instructions="You are Alex at Acme.",
    skills=Path(__file__).parent / "skills",
)

Run python skills_agent.py, then say (or send) "I just got my new card — how do I activate it?" and watch a cosmo_sdk_load_skill tool call appear in the event stream before the agent answers with the skill's flow. Skills explains the tiering.

Guard tools with hooks

hooks_agent.py registers one hook per seam — inject context at session start, deny destructive tools by name pattern, observe every outcome, log the exit reason:

from cosmo_ai import RealtimeClient, hooks
from cosmo_ai.hooks import PreToolUseContext, PreToolUseResult, SessionStartContext, SessionStartResult

@hooks.session_start
def add_context(ctx: SessionStartContext) -> SessionStartResult:
    return SessionStartResult(additional_context="Be concise and warm.")

@hooks.pre_tool_use(matcher="delete_*")
def block_deletes(ctx: PreToolUseContext) -> PreToolUseResult:
    return PreToolUseResult(permission="deny", reason="destructive tools are disabled")

agent = client.agent(instructions="You are Alex.", hooks=[add_context, block_deletes])

Run python hooks_agent.py. Any tool whose name matches delete_* is denied before its handler runs — the agent is told why and adjusts, and your post_tool_use hook sees the denied outcome. Hooks covers matchers and fold rules.

Compose production guardrails

production_hooks.py composes the patterns from Add production guardrails into one session: a pre_tool_use hook redacts emails and phone numbers before a case note reaches its handler, a post_tool_use hook appends every client-tool outcome to a JSONL audit trail, a wrap-up tool raises while the call's goal is unmet so the model recovers instead of hanging up, and a session_end hook queues follow-up for a call that ends incomplete on any exit path.

Run python production_hooks.py. The driver first asks the agent to end the call with nothing documented (the wrap-up refusal comes back as a tool error), then sends a note carrying an email and phone number and wraps up — the filed note and the audit trail print with the PII already redacted.

Wire tools from an MCP server

mcp_agent.py gets its tools from a local stdio MCP server declared in mcp.json — no per-tool code at all:

agent = client.agent(
    instructions="You can use the connected MCP tools to help the user.",
    mcp=Path(__file__).parent / "mcp.json",
)

Install the extra first (pip install "cosmo-ai-sdk[mcp]"), then run python mcp_agent.py. The SDK launches the server, lists its tools, declares them to the agent, and proxies calls back — ask for something a listed tool can do and the round-trip shows up in the printed events. MCP covers the config format.

Dial an outbound phone call

outbound_call.py starts a session with no participants, then brings a real phone into the room. No local microphone is involved — the human on this call is the phone:

async with agent.start() as session:
    print(f"Session live (id={session.session_id}); dialing {phone_number}…")
    try:
        dial = await session.dial(phone_number)
    except DialError as exc:
        raise SystemExit(f"Dial rejected ({exc.code}): {exc.message}")
    print(f"Dial queued: {dial.dial_id}")

Unlike the other sends, dial is an authenticated REST call: it requires an API key from a workspace with phone calls enabled, and the call counts against the workspace's calling limits. Run it as COSMO_DIAL_TO=+14155550199 python outbound_call.py (an E.164 number you control), answer the phone, and the transcript of your own call prints to the terminal. Without telephony access the script exits with a DialError such as phone_calls_disabled — see Telephony for enabling it.

Next steps

• Voice CLI — the same event loop with real microphone and speaker audio
• Tools, Skills, Hooks, MCP — each capability in depth
• Python reference — the full client.agent(...) parameter list