claudestream

Drive Claude Code from Python: run it as a subprocess and read its output as typed events, with async and sync sessions, sandbox policies, and tools you define in Python

Install

uv pip install claudestream

Quick start

One-shot ask

from claudestream import SessionConfig, SyncSession

config = SessionConfig(model="sonnet", profile="default")
with SyncSession(config) as session:
    result = session.ask("What is 2 + 2?")
    print(result.text)

Streaming events

from claudestream import SessionConfig, SyncSession, AssistantText, ToolUse, Result

config = SessionConfig(model="sonnet", profile="default")
with SyncSession(config) as session:
    for event in session.send("List the files in the current directory"):
        if isinstance(event, AssistantText):
            print(event.text, end="")
        elif isinstance(event, ToolUse):
            print(f"\n[tool: {event.name}]")
        elif isinstance(event, Result):
            print(f"\n(cost: ${event.total_cost_usd:.4f})")

Async session

import asyncio
from claudestream import SessionConfig, AsyncSession, AssistantText

async def main():
    config = SessionConfig(model="sonnet", profile="default")
    async with AsyncSession(config) as session:
        async for event in session.send("Hello!"):
            if isinstance(event, AssistantText):
                print(event.text, end="")

asyncio.run(main())

Multi-turn conversation

from claudestream import SessionConfig, SyncSession, AssistantText

config = SessionConfig(model="sonnet", profile="default")
with SyncSession(config) as session:
    for event in session.send("Remember that my name is Alice."):
        if isinstance(event, AssistantText):
            print(event.text, end="")
    print()
    for event in session.send("What is my name?"):
        if isinstance(event, AssistantText):
            print(event.text, end="")

Custom tools

Define tools with the @tool decorator. claudestream auto-generates JSON Schema from type hints and serves them via MCP.

from claudestream import tool, collect_tools, SessionConfig, SyncSession, AssistantText

@tool("my_server")
def lookup_weather(city: str, units: str = "celsius") -> str:
    """Look up current weather for a city.

    Args:
        city: City name to look up.
        units: Temperature units, celsius or fahrenheit.
    """
    return f"22 degrees {units} in {city}"

config = SessionConfig(
    model="sonnet",
    profile="default",
    tools=[lookup_weather._tool],
)
with SyncSession(config) as session:
    for event in session.send("What's the weather in Paris?"):
        if isinstance(event, AssistantText):
            print(event.text, end="")

Agents

Agents are JSON-defined configurations with prompt templates, tool schemas, sandbox policies, and budget limits.

from claudestream import (
    load_agent, invoke_agent_sync, SessionConfig, AssistantText,
)

agent = load_agent("code_reviewer")  # loads .claudestream/agents/code_reviewer.agent.json
config = SessionConfig(model="sonnet", profile="default")

with invoke_agent_sync(agent, config, variables={"file": "main.py"}) as session:
    for event in session.send("Review this file"):
        if isinstance(event, AssistantText):
            print(event.text, end="")

Sandbox policies

Restrict which tools Claude can use and which paths it can write to.

from claudestream import create_sandbox, SessionConfig, SyncSession

sandbox = create_sandbox(
    tools=["Read", "Bash"],
    write_paths=["/home/user/project"],
)
config = SessionConfig(model="sonnet", profile="default", sandbox=sandbox)

with SyncSession(config) as session:
    result = session.ask("Read the README and summarize it")
    print(result.text)

CLI

Configuration

Sandbox fields

Dependencies

Modules

• claudestream (claudestream/__init__.py): Drive Claude Code from Python: run it as a subprocess and read its output as typed events, with async and sync sessions, sandbox policies, and tools you define in Python.
• claudestream._agent (claudestream/_agent.py): Agent definition loader and budget enforcement for Claude Code sessions, with sync and async context managers for invoking agents.
• claudestream._agent_schema (claudestream/_agent_schema.py)
• claudestream._async_session (claudestream/_async_session.py): Async session manager for the Claude Code stream-json protocol, handling process lifecycle, event parsing, and permission callbacks.
• claudestream._cli (claudestream/_cli.py): Command-line interface entry point for claudestream, providing send, listen, and agent commands for interacting with Claude Code.
• claudestream._color (claudestream/_color.py): ANSI color output support with automatic TTY detection, NO_COLOR environment variable compliance, and a reusable Colorizer class.
• claudestream._options (claudestream/_options.py): Option structs for configuring claudestream sessions, covering session resolution, debug, MCP, plugins, stream output, process limits, budget, tool schema, and the unified SessionConfig.
• claudestream._process (claudestream/_process.py): Subprocess management for launching and monitoring the Claude Code CLI process, including graceful shutdown and atexit cleanup.
• claudestream._protocol (claudestream/_protocol.py): NDJSON protocol layer that reads raw Claude Code stream-json output lines and decodes them into typed Event objects for consumption.
• claudestream._sync_session (claudestream/_sync_session.py): Synchronous session wrapper that bridges the async Claude Code stream-json protocol to a blocking iterator-based interface.
• claudestream._tools (claudestream/_tools.py): Tool registration API providing the Tool struct and a decorator for defining user tools that are served via MCP to Claude Code.
• claudestream.events (claudestream/events.py): Typed event dataclasses for every Claude Code stream output event, including assistant messages, tool use, permissions, and results.
• claudestream.messages (claudestream/messages.py): Typed message structs for all Claude Code stream input messages, including user prompts, tool results, and permission responses.
• claudestream.policy (claudestream/policy.py): Sandbox and permission policy types for Claude Code sessions, defining allow, deny, and approval rules for tool execution requests.

Project layout

claudestream/
├── __init__.py
├── _agent.py
├── _agent_schema.py
├── _async_session.py
├── _cli.py
├── _color.py
├── _options.py
├── _process.py
├── _protocol.py
├── _sync_session.py
├── _tools.py
├── events.py
├── messages.py
└── policy.py

License

MIT