CLI assistant

"Talk to my computer" — declare client tools in Python so the agent can run shell commands, read clipboard content, and interact with the desktop.

The CLI Assistant gives you a voice-controlled interface to your local machine. You speak a request; the agent calls client tools that run in your Python process — shell commands, clipboard reads, file listings — and the results flow back to the agent automatically.

Architecture

Voice input (mic)
    ↓
cosmo-ai-sdk session (Cosmo agent)
    ↓ client-tool invocations over the transport RPC bridge
Python process (your @tool handlers)
    ↑ handler return values become the tool results

Client tools are first-class: the @tool decorator derives the model-facing schema from a Pydantic input model, the SDK validates the model's arguments before your handler runs, and the handler's return value is delivered back to the agent — no manual dispatch or send_text plumbing.

System prompt

You are a command-line assistant running on the user's computer.

You have access to local tools: run_shell_command, read_clipboard, list_directory.
When the user asks you to run something, call run_shell_command with the exact command.
Keep output concise — summarise long outputs rather than quoting them fully.

Do not run destructive commands (rm -rf, format, etc.) without explicit user confirmation.

Python implementation

"""
cli_assistant.py — voice-controlled local assistant.

Usage:
    pip install cosmo-ai-sdk
    export COSMO_API_KEY=cosmo_...
    python cli_assistant.py
"""

from __future__ import annotations

import asyncio
import os
import subprocess
import sys
from typing import Any

from pydantic import BaseModel, Field

from cosmo_ai import (
    CosmoRealtime,
    RealtimeReady,
    RealtimeSessionEnded,
    RealtimeToolCall,
    RealtimeToolResult,
    RealtimeTranscriptDelta,
    tool,
)

SYSTEM_PROMPT = """You are a command-line assistant running on the user's computer.
Keep output concise — summarise long outputs rather than quoting them fully.
Do not run destructive commands without explicit user confirmation."""

# ── Local tools ──────────────────────────────────────────────────────────────

class ShellInput(BaseModel):
    command: str = Field(description="The exact shell command to run")


@tool
async def run_shell_command(input: ShellInput) -> dict[str, Any]:
    """Run a shell command on the user's machine and return its output."""
    try:
        result = subprocess.run(
            input.command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=30,
        )
        output = result.stdout.strip() or result.stderr.strip()
        return {"output": output[:2000] if output else "(no output)"}
    except subprocess.TimeoutExpired:
        return {"error": "command timed out after 30 seconds"}


class ClipboardInput(BaseModel):
    pass


@tool
async def read_clipboard(input: ClipboardInput) -> dict[str, Any]:
    """Read the current clipboard contents."""
    try:
        # macOS
        result = subprocess.run(["pbpaste"], capture_output=True, text=True)
        return {"clipboard": result.stdout or "(clipboard is empty)"}
    except FileNotFoundError:
        # Linux (xclip)
        try:
            result = subprocess.run(
                ["xclip", "-selection", "clipboard", "-o"],
                capture_output=True,
                text=True,
            )
            return {"clipboard": result.stdout or "(clipboard is empty)"}
        except FileNotFoundError:
            return {"error": "clipboard tool not available"}


class ListDirectoryInput(BaseModel):
    path: str = Field(default=".", description="Directory to list")


@tool
async def list_directory(input: ListDirectoryInput) -> dict[str, Any]:
    """List the entries in a directory."""
    entries = os.listdir(input.path)
    return {"entries": sorted(entries)[:50]}


# ── Main ─────────────────────────────────────────────────────────────────────

async def run(api_key: str) -> None:
    async with CosmoRealtime(api_key=api_key) as client:
        agent = client.agent(
            instructions=SYSTEM_PROMPT,
            tools=[run_shell_command, read_clipboard, list_directory],
        )
        async with agent.start() as session:
            print("Connecting…")

            async def stop_on_enter() -> None:
                loop = asyncio.get_event_loop()
                await loop.run_in_executor(None, sys.stdin.readline)
                print("Ending session…")
                await session.end()

            stopper = asyncio.create_task(stop_on_enter())

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

            async for event in session:
                if isinstance(event, RealtimeReady):
                    print("Ready. Speak to your computer. Press Enter to stop.")
                elif isinstance(event, RealtimeTranscriptDelta):
                    if event.is_final:
                        print(f"  [{event.role.value}] {event.text}", flush=True)
                elif isinstance(event, RealtimeToolCall):
                    print(f"  [tool] {event.name}", flush=True)
                elif isinstance(event, RealtimeToolResult):
                    status = "ok" if event.ok else "error"
                    print(f"  [result] {status} — {event.summary or ''}", flush=True)
                elif isinstance(event, RealtimeSessionEnded):
                    print(f"Session ended: {event.reason}")

            await stopper


def main() -> None:
    api_key = os.environ.get("COSMO_API_KEY", "")
    if not api_key:
        sys.exit("Set COSMO_API_KEY")
    asyncio.run(run(api_key=api_key))


if __name__ == "__main__":
    main()

Usage

export COSMO_API_KEY=cosmo_...
python cli_assistant.py

Then speak:

"Run ls in my home directory"
  [user] Run ls in my home directory »
  [tool] run_shell_command
  [assistant] Your home directory contains: Desktop, Documents, Downloads…

"What's in my clipboard?"
  [user] What's in my clipboard? »
  [tool] read_clipboard
  [assistant] Your clipboard contains: https://example.com/…

TypeScript (Node)

For a Node.js CLI assistant, build the same tools with the tool() builder and drive the session from stdin:

import { RealtimeClient } from 'cosmo-ai';
import { tool } from 'cosmo-ai/tool';
import { zodInput } from 'cosmo-ai/tool/zod';
import { z } from 'zod/v4';
import { execSync } from 'node:child_process';
import * as readline from 'node:readline';

const runShellCommand = tool({
  name: 'run_shell_command',
  description: 'Run a shell command on the user’s machine and return its output',
  input: zodInput(
    z.object({ command: z.string().describe('The exact shell command to run') }),
  ),
  handler: async ({ command }) => {
    try {
      return {
        output: execSync(command, { timeout: 30_000, encoding: 'utf8' }).slice(0, 2000),
      };
    } catch (e: unknown) {
      const err = e as { message?: string };
      return { error: err.message ?? 'command failed' };
    }
  },
});

const client = new RealtimeClient({
  baseUrl: 'https://app.askcosmo.ai',
  apiKey: process.env.COSMO_API_KEY,
});

const agent = client.agent({
  instructions: SYSTEM_PROMPT,
  tools: [runShellCommand],
});
const session = await agent.start();

session.on('transcript', (event) => {
  if (event.isFinal) {
    console.log(`[${event.role}]`, event.text);
  }
});

session.on('tool_call', (event) => {
  console.log('tool called:', event.name);
});

// Read from stdin until Ctrl-C.
const rl = readline.createInterface({ input: process.stdin });
rl.on('line', (line) => {
  if (line.trim()) void session.sendText(line.trim());
});
process.on('SIGINT', () => {
  void session.end().finally(() => process.exit(0));
});

An apiKey client requires baseUrl and belongs in a server-side or local process like this one. If you are embedding the assistant in an app you distribute to end users, mint a short-lived token on your backend instead and construct the client with { token }.

Safety considerations

Client tools give the agent the ability to execute arbitrary shell commands. Before deploying:

Validate or allowlist the command argument in run_shell_command.
Add a pre-tool-use hook matching run_shell_command that denies destructive commands — hooks run before every handler, no matter what the model decides.
Run the process with minimal OS permissions (not as root).
Consider sandboxing with firejail, containers, or a separate VM.
Log all tool calls and results for audit.

Adding server tools

This is a local-tools-first session — by default it runs with your client tools only. To let the agent search the web too, add the typed, zero-config opt-in for the server-executed tool:

from cosmo_ai import WebSearchTool

agent = client.agent(
    instructions=SYSTEM_PROMPT,
    tools=[run_shell_command, read_clipboard, list_directory,
           WebSearchTool()],
)

Next steps

Build a Python voice CLI — simpler baseline without local tools
Tools — the full client / background / server tool taxonomy
Hooks — deny or rewrite tool calls before they execute
Debugging — log tool calls with session IDs

CLI assistant

On this page