Updated July 29, 2026

On this page

How claudestream's 4-layer architecture (process, protocol, session, CLI) turns a Claude Code subprocess into typed async events with permissions.

#Architecture Guide

claudestream wraps the Claude Code CLI's stream-json protocol in a four-layer Python SDK. Each layer adds structure on top of the one below it: Process spawns the subprocess, Protocol decodes its output, Session manages conversation state, and CLI exposes it all as shell commands.

#The four layers

#Process (bottom)

The process layer spawns and manages the Claude Code subprocess, mapping every session option to its corresponding CLI flag and handling lifecycle events including startup, graceful shutdown with a 3-stage sequence, and atexit cleanup for orphan prevention.

#`claudestream._process`

Subprocess management for launching and monitoring the Claude Code CLI process, including graceful shutdown and atexit cleanup.

#`_kill_active_children`

python

def _kill_active_children()

atexit handler: SIGTERM all tracked child processes.

#`find_binary`

python

def find_binary(binary: str | None=None) -> str

Locate the claude CLI binary.

Args:

binary: Explicit path to claude binary. If None, searches PATH.

Returns:

Absolute path to the claude binary.

Raises:

FileNotFoundError: If claude is not found.

#`_version_lt`

python

def _version_lt(a: str, b: str) -> bool

Return True if version a < version b (semver comparison).

#`check_version`

python

async def check_version(binary: str, *, timeout: float=2.0) -> str | None

Check claude CLI version. Logs warning if below minimum. Returns version string or None.

#`ProcessConfig`

Configuration for spawning a Claude Code subprocess.

#`build_argv`

python

def build_argv(self) -> list[str]

Build the full command-line argument list from the flag registry.

#`ProcessManager`

Manages a Claude Code subprocess lifecycle.

#`stdin`

python

def stdin(self) -> asyncio.StreamWriter

#`stdout`

python

def stdout(self) -> asyncio.StreamReader

#`is_alive`

python

def is_alive(self) -> bool

#`stderr_lines`

python

def stderr_lines(self) -> list[str]

#`_drain_stderr`

python

async def _drain_stderr(self) -> None

Read stderr line-by-line to prevent pipe buffer deadlock.

#`start`

python

async def start(self) -> None

Spawn the claude subprocess.

#`close`

python

async def close(self) -> None

Graceful shutdown: close stdin -> wait -> SIGTERM -> wait -> SIGKILL.

#`kill`

python

async def kill(self) -> None

Immediate kill.

ProcessConfig is a frozen struct that maps every session option to its CLI flag equivalent. Its build_argv() method produces the full argument list, including the hardcoded --output-format stream-json --input-format stream-json that enables the protocol. A declarative flag registry (_FLAG_REGISTRY) drives the mapping: each entry is a (field_name, cli_flag, style) tuple where style is "value", "bool", or "list".

ProcessManager owns the subprocess lifecycle. On start(), it spawns the process with piped stdin/stdout/stderr, registers it in a module-level _ACTIVE_CHILDREN set (cleaned up by an atexit handler), and launches a background task to drain stderr. On close(), it follows a three-stage shutdown sequence: close stdin, wait for exit, SIGTERM with timeout, then SIGKILL.

#Protocol (middle-lower)

The protocol layer converts raw NDJSON lines into typed Python Event objects and serializes outbound Message objects back to NDJSON, providing the bidirectional codec between the subprocess I/O streams and the SDK's typed event system.

#`claudestream._protocol`

NDJSON protocol layer that reads raw Claude Code stream-json output lines and decodes them into typed Event objects for consumption.

#`parse_content_block`

python

def parse_content_block(raw: dict) -> ContentBlock

Parse a single content block dict into the correct typed block.

#`parse_usage`

python

def parse_usage(raw: dict | None) -> Usage | None

Parse a usage dict into Usage, or return None.

#`parse_event`

python

def parse_event(raw: dict) -> Event

Map a raw JSON dict to the correct typed Event Struct.

#`_resolve_path`

python

def _resolve_path(path: str, cwd: str | None) -> str

Resolve a file path to absolute, using cwd if the path is relative.

#`_derive_file_events`

python

def _derive_file_events(block: ToolUseBlock, event: AssistantMessage, cwd: str | None) -> list[Event]

Derive FileWrite/FileEdit events from a file-modifying ToolUseBlock.

#`flatten_event`

python

def flatten_event(event: Event, cwd: str | None=None) -> list[Event]

Expand an event into convenience events (one per content block).

Args:

event: The event to flatten.
cwd: Working directory for resolving relative paths in file-tracking events.

#`read_events`

python

async def read_events(stream: asyncio.StreamReader) -> AsyncIterator[Event]

Async generator that reads NDJSON lines and yields parsed Events.

#`write_message`

python

async def write_message(stream: asyncio.StreamWriter, msg: Writable) -> None

Serialize a message to NDJSON and write it to the stream.

The protocol layer converts between raw NDJSON lines and typed Python objects.

Reading (subprocess to SDK): read_events() is an async generator that reads lines from an asyncio.StreamReader, JSON-decodes each line, and calls parse_event(). The parser dispatches on the type field ("system", "assistant", "user", "result", "control_request", etc.) and further on subtype fields to construct the correct Event subclass. Unrecognized types become UnknownEvent for forward compatibility.

Writing (SDK to subprocess): write_message() takes any Writable message (a union of UserMessage, AllowPermission, DenyPermission, and several others), calls its to_dict() method, serializes to JSON, appends a newline, and writes to the asyncio.StreamWriter.

Flattening: flatten_event() expands compound events into convenience events. An AssistantMessage with three content blocks (text, tool_use, thinking) becomes three separate events (AssistantText, ToolUse, Thinking). A ToolResultMessage is expanded into individual ToolResult events. Tool use blocks for Write, Edit, and MultiEdit also generate derived FileWrite/FileEdit events for file-tracking. Non-compound events pass through as single-element lists.

#Session (middle-upper)

The session layer manages turn-based conversation state on top of the protocol, combining process lifecycle, event parsing, sandbox-based permission interception, MCP tool serving, budget tracking, and callback dispatch into a single context manager.

#`claudestream._async_session`

Async session manager for the Claude Code stream-json protocol, handling process lifecycle, event parsing, and permission callbacks.

#`ClaudeStreamError`

Raised when the Claude Code subprocess fails.

#`AsyncSession`

Async session managing a Claude Code subprocess.

Usage::

async with AsyncSession(model="sonnet") as session: async for event in session.send("hello"): print(event)

#`_enrich_flattened`

python

def _enrich_flattened(self, evt: Event) -> Event

Correlate flattened tool events: record ToolUse names and stamp the matching name onto a ToolResult. Returns the (possibly replaced) event.

#`_build_process_config`

python

def _build_process_config(self) -> ProcessConfig

Build a ProcessConfig from the stored SessionConfig.

This is the single mapping point between the user-facing config and the subprocess CLI flags.

#`session_id`

python

def session_id(self) -> str | None

#`model_name`

python

def model_name(self) -> str | None

#`tools`

python

def tools(self) -> list[str]

#`claude_version`

python

def claude_version(self) -> str | None

#`last_result`

python

def last_result(self) -> Result | None

#`turn_count`

python

def turn_count(self) -> int

#`total_tokens`

python

def total_tokens(self) -> int

#`total_cost_usd`

python

def total_cost_usd(self) -> float

#`stderr_lines`

python

def stderr_lines(self) -> list[str]

#`sandbox`

python

def sandbox(self) -> Sandbox | None

#`user_tools`

python

def user_tools(self) -> list[Tool]

#`is_alive`

python

def is_alive(self) -> bool

#`active_turn`

python

def active_turn(self) -> bool

#`cancelled`

python

def cancelled(self) -> bool

#`files_modified`

python

def files_modified(self) -> set[str]

All files written or edited during this session (absolute paths, deduplicated).

Note: Only tracks files modified via Write, Edit, and MultiEdit tools. Files modified via Bash tool calls are not tracked.

#`process_pid`

python

def process_pid(self) -> int | None

#`cwd`

python

def cwd(self) -> str

#`mcp_servers`

python

def mcp_servers(self) -> list[str]

#`permission_mode`

python

def permission_mode(self) -> str

#`restart_count`

python

def restart_count(self) -> int

#`config`

python

def config(self) -> SessionConfig

#`_start`

python

async def _start(self) -> None

Start the subprocess and complete the MCP handshake if tools are registered.

When SDK MCP tools are registered, the full handshake is completed before returning so tools are ready by the time the first send() is called:

Send InitializeRequest -> read ControlResponse
Send McpSetServers -> read ControlResponse
Read and respond to MCP handshake messages (initialize, notifications/initialized, tools/list)

SystemInit (if received during handshake) is stored and drained on first send().

#`_read_control_response`

python

async def _read_control_response(self, timeout: float=10.0) -> ControlResponse

Read events from stdout until a ControlResponse is received.

Any non-ControlResponse events encountered are stored in _startup_events.

#`_run_mcp_handshake`

python

async def _run_mcp_handshake(self, timeout: float=10.0) -> None

Complete the MCP protocol handshake for ALL registered MCP servers.

Each server goes through: initialize -> notifications/initialized -> tools/list. We must wait for every server's tools/list before returning, otherwise send() will write a UserMessage to stdin before the CLI finishes handshaking remaining servers, causing a protocol-level hang.

#`close`

python

async def close(self) -> None

Shut down the session and kill the subprocess.

#`cancel`

python

async def cancel(self, force: bool=False) -> None

Cancel the current operation.

Args:

force: If False, close stdin (graceful). If True, terminate subprocess.

#`send`

python

async def send(self, prompt: str | list, *, raw: bool=False) -> AsyncIterator[Event]

Send a message and yield events until the turn completes.

Args:

prompt: The message to send. Can be a plain string or a list of

content blocks (dicts) for multimodal input.

raw: If True, yield raw protocol events (AssistantMessage,

ToolResultMessage). If False (default), yield flattened convenience events (AssistantText, ToolUse, etc.).

Yields:

Event objects until a Result event is received.

Raises:

RuntimeError: If called while a previous turn is still active.
ClaudeStreamError: If the subprocess dies unexpectedly.

#`ask`

python

async def ask(self, prompt: str | list) -> AskResult

Send a prompt and return the complete response text with metadata.

#`_read_turn`

python

async def _read_turn(self, *, raw: bool, _health_timeout: float=30.0) -> AsyncIterator[Event]

Read events for a single turn until Result is received.

#`_liveness_probe`

python

async def _liveness_probe(self) -> None

Check if the subprocess is still alive.

Raises:

ClaudeStreamError: If the process has died.

#`_restart_subprocess`

python

async def _restart_subprocess(self) -> None

Kill the stuck subprocess and restart with --resume to preserve session.

#`_handle_permission`

python

async def _handle_permission(self, request: PermissionRequest) -> bool

Apply sandbox rules to a permission request. Returns True if handled.

#`_handle_mcp_request`

python

async def _handle_mcp_request(self, request: McpRequest) -> bool

Handle an MCP JSON-RPC request. Returns True if handled.

#`_resolve_control`

python

def _resolve_control(self, event: ControlResponse) -> bool

Resolve the pending future for a control response. Returns True if matched.

#`_fail_pending_controls`

python

def _fail_pending_controls(self, message: str) -> None

Fail every pending control future with ClaudeStreamError and clear the registry.

#`_control_request`

python

async def _control_request(self, subtype: str, payload: dict | None=None, *, timeout: float=30.0) -> dict

Issue a control request and await its correlated control_response.

During an active turn the turn loop resolves the future (no stdout read happens here). Between turns this drives its own scoped stdout read loop, buffering unrelated events for the next turn.

Returns the inner response dict on success. Raises ClaudeStreamError on process death, CLI error response, or timeout.

#`_read_control_result`

python

async def _read_control_result(self, request_id: str, future: asyncio.Future, timeout: float) -> dict

Read stdout until the matching control_response resolves the future.

Non-matching events are buffered into _startup_events for the next turn. Respects the per-read health timeout and the overall operation timeout; raises ClaudeStreamError on EOF.

#`_check_thresholds`

python

def _check_thresholds(self) -> list

Check all threshold lists and return BudgetThreshold events for newly-crossed thresholds.

#`_write_cost_log`

python

def _write_cost_log(self, result) -> None

Append a JSONL line to the cost log file if configured.

#`on`

python

def on(self, event_type: type[Event], handler: Callable[[Any], None]) -> None

The callback fires during iteration, before the event is yielded.

#`on_turn_complete`

python

def on_turn_complete(self, hook: Callable) -> None

Hook signature: async def hook(session, result) or def hook(session, result). Hooks run in registration order. Errors are logged but do not propagate.

#`on_error`

python

def on_error(self, hook: Callable) -> None

Hook signature: async def hook(session, exception) or def hook(session, exception). Hooks run in registration order. Errors are logged but do not propagate.

#`on_close`

python

def on_close(self, hook: Callable) -> None

Hook signature: async def hook(session) or def hook(session). Hooks run in registration order. Errors are logged but do not propagate.

#`_fire_hooks`

python

async def _fire_hooks(self, hooks: list[Callable], *args: Any) -> None

Fire a list of hooks with the given arguments, logging and swallowing errors.

#`respond_allow`

python

async def respond_allow(self, request_id: str, updated_input: dict, *, updated_permissions: list[dict] | None=None) -> None

Allow a permission request that was surfaced to the consumer.

updated_permissions optionally carries permission-rule updates to apply alongside the allow; it is omitted from the wire frame when None.

#`respond_deny`

python

async def respond_deny(self, request_id: str, message: str='Denied by user') -> None

Deny a permission request that was surfaced to the consumer.

#`respond_dialog`

python

async def respond_dialog(self, request_id: str, result: Any) -> None

Complete a user dialog request with the user's chosen result.

result is transported opaquely; its shape is defined per dialog_kind.

#`respond_dialog_cancelled`

python

async def respond_dialog_cancelled(self, request_id: str) -> None

Cancel a user dialog request; the CLI applies the dialog's default behavior.

#`interrupt`

python

async def interrupt(self, *, timeout: float=30.0) -> list[str]

Interrupt the running turn.

Callable while a turn is active -- that is its purpose. Returns the list of user messages the CLI still had queued (empty on older CLIs that omit the field).

#`set_permission_mode`

python

async def set_permission_mode(self, mode: str) -> None

Change the permission mode mid-session.

The mode string is passed through unvalidated; the CLI rejects unknown modes with an error response. On success the permission_mode property is updated to reflect the new mode.

#`set_model`

python

async def set_model(self, model: str | None) -> None

Switch the model mid-session.

Passing None omits the model field, resetting the CLI to its default. On success the model_name property is updated (None means the model is unknown until the next SystemInit reports it).

#`get_context_usage`

python

async def get_context_usage(self, *, timeout: float=30.0) -> ContextUsage

Query the model's current context-window usage.

Raises ClaudeStreamError if the CLI response omits totalTokens/maxTokens.

#`claudestream._sync_session`

Synchronous session wrapper that bridges the async Claude Code stream-json protocol to a blocking iterator-based interface.

#`SyncSession`

Synchronous session managing a Claude Code subprocess.

Wraps AsyncSession by running it on a dedicated event loop thread.

Usage::

config = SessionConfig(model="sonnet", profile="default") with SyncSession(config) as session: for event in session.send("hello"): print(event)

#`_run_loop`

python

def _run_loop(self) -> None

Target for the event loop thread.

#`_ensure_loop`

python

def _ensure_loop(self) -> asyncio.AbstractEventLoop

Start the event loop thread if not already running.

#`_run_coro`

python

def _run_coro(self, coro)

Run a coroutine on the event loop thread and wait for the result.

#`close`

python

def close(self) -> None

Shut down the session, subprocess, and event loop thread.

#`session_id`

python

def session_id(self) -> str | None

#`model_name`

python

def model_name(self) -> str | None

#`tools`

python

def tools(self) -> list[str]

#`claude_version`

python

def claude_version(self) -> str | None

#`last_result`

python

def last_result(self) -> Result | None

#`files_modified`

python

def files_modified(self) -> set[str]

All files written or edited during this session (absolute paths, deduplicated).

#`stderr_lines`

python

def stderr_lines(self) -> list[str]

#`turn_count`

python

def turn_count(self) -> int

#`total_tokens`

python

def total_tokens(self) -> int

#`total_cost_usd`

python

def total_cost_usd(self) -> float

#`sandbox`

python

def sandbox(self) -> Sandbox | None

#`user_tools`

python

def user_tools(self) -> list[Tool]

#`is_alive`

python

def is_alive(self) -> bool

#`active_turn`

python

def active_turn(self) -> bool

#`cancelled`

python

def cancelled(self) -> bool

#`process_pid`

python

def process_pid(self) -> int | None

#`cwd`

python

def cwd(self) -> str

#`mcp_servers`

python

def mcp_servers(self) -> list[str]

#`permission_mode`

python

def permission_mode(self) -> str

#`config`

python

def config(self) -> SessionConfig

#`cancel`

python

def cancel(self, force: bool=False) -> None

Cancel the current operation.

Args:

force: If False, close stdin (graceful). If True, terminate subprocess.

#`ask`

python

def ask(self, prompt: str | list) -> AskResult

Send a prompt and return the complete response text with metadata.

#`send`

python

def send(self, prompt: str | list, *, raw: bool=False) -> Iterator[Event]

Send a message and yield events until the turn completes.

Args:

prompt: The message to send. Can be a plain string or a list of

content blocks (dicts) for multimodal input.

raw: If True, yield raw protocol events. If False, yield flattened events.

Yields:

Event objects until a Result event is received.

#`on`

python

def on(self, event_type: type[Event], handler: Callable[[Any], None]) -> None

#`on_turn_complete`

python

def on_turn_complete(self, hook: Callable) -> None

Hook signature: def hook(session, result). The session argument is this SyncSession instance (not the underlying AsyncSession).

#`on_error`

python

def on_error(self, hook: Callable) -> None

Hook signature: def hook(session, exception). The session argument is this SyncSession instance (not the underlying AsyncSession).

#`on_close`

python

def on_close(self, hook: Callable) -> None

Hook signature: def hook(session). The session argument is this SyncSession instance (not the underlying AsyncSession).

#`respond_allow`

python

def respond_allow(self, request_id: str, updated_input: dict, *, updated_permissions: list[dict] | None=None) -> None

Allow a permission request, optionally applying permission-rule updates.

#`respond_deny`

python

def respond_deny(self, request_id: str, message: str='Denied by user') -> None

Deny a permission request.

#`respond_dialog`

python

def respond_dialog(self, request_id: str, result: Any) -> None

Complete a user dialog request with the user's chosen result.

#`respond_dialog_cancelled`

python

def respond_dialog_cancelled(self, request_id: str) -> None

Cancel a user dialog request; the CLI applies the dialog's default behavior.

#`interrupt`

python

def interrupt(self, *, timeout: float=30.0) -> list[str]

Interrupt the running turn. Returns any still-queued user messages.

#`set_permission_mode`

python

def set_permission_mode(self, mode: str) -> None

Change the permission mode mid-session.

#`set_model`

python

def set_model(self, model: str | None) -> None

Switch the model mid-session. None resets to the CLI default.

#`get_context_usage`

python

def get_context_usage(self, *, timeout: float=30.0) -> ContextUsage

Query the model's current context-window usage.

AsyncSession is the primary implementation. It combines the process and protocol layers with conversation state, permission handling, MCP tool serving, budget tracking, and lifecycle hooks.

SyncSession is a thin wrapper that runs an AsyncSession on a dedicated event loop thread. It bridges the async iterator to a blocking queue.Queue: a background coroutine drains the async iterator and puts events on the queue; the sync send() method polls the queue with timeouts. All property access and method calls are forwarded via run_coroutine_threadsafe.

#CLI (top)

The CLI layer exposes the SDK as shell commands built with strictcli, providing 8 commands for sending prompts, streaming tokens, debugging protocol events, running interactive sessions, and managing agent definitions.

#`claudestream._cli`

Command-line interface entry point for claudestream, providing send, listen, and agent commands for interacting with Claude Code.

#`_get_version`

python

def _get_version() -> str

Read version from pyproject.toml (editable installs) or fall back to package metadata.

#`_resolve_prompt`

python

def _resolve_prompt(prompt: str, stdin: bool, color: Colorizer) -> str | int

Resolve prompt from argument or stdin. Returns the prompt string, or 1 on error.

#`_build_config`

python

def _build_config(model: str, profile: str, cwd: str='', skip_permissions: bool=False, system_prompt: str='', resume: str='', from_pr: str='') -> SessionConfig

Build a SessionConfig from common CLI flags.

#`_run_with_session`

python

def _run_with_session(config: SessionConfig, handler: Any, color: Colorizer) -> int | None

Run handler(session) inside a SyncSession context with standard error handling.

#`_stream_events`

python

def _stream_events(session: SyncSession, prompt: str, footer: bool, color: Colorizer) -> None

Shared streaming event loop used by cmd_stream and cmd_agent_run.

#`cmd_send`

python

def cmd_send(ctx, prompt: str='', model: str='', profile: str='', cwd: str='', raw: bool=False, json_output: bool=False, skip_permissions: bool=False, footer: bool=True, system_prompt: str='', stdin: bool=False, color: bool=True, resume: str='', from_pr: str='') -> int | None

#`cmd_stream`

python

def cmd_stream(ctx, prompt: str='', model: str='', profile: str='', cwd: str='', skip_permissions: bool=False, footer: bool=True, system_prompt: str='', stdin: bool=False, color: bool=True, resume: str='', from_pr: str='') -> int | None

#`cmd_events`

python

def cmd_events(ctx, prompt: str='', model: str='', profile: str='', cwd: str='', skip_permissions: bool=False, footer: bool=True, system_prompt: str='', stdin: bool=False, color: bool=True, resume: str='', from_pr: str='') -> int | None

#`cmd_repl`

python

def cmd_repl(ctx, model: str, profile: str, cwd: str='', skip_permissions: bool=False, footer: bool=True, system_prompt: str='', color: bool=True, resume: str='', from_pr: str='') -> None

#`cmd_agent_run`

python

def cmd_agent_run(ctx, definition: str, prompt: str, var: list[str], model: str, profile: str, cwd: str='', footer: bool=True, color: bool=True) -> int | None

#`cmd_agent_list`

python

def cmd_agent_list(ctx, cwd: str='') -> int | None

#`cmd_agent_info`

python

def cmd_agent_info(ctx, name: str) -> int | None

#`cmd_agent_validate`

python

def cmd_agent_validate(ctx, name: str) -> int | None

#`cmd_ask`

python

def cmd_ask(ctx, prompt: str='', model: str='', profile: str='', cwd: str='', skip_permissions: bool=False, system_prompt: str='', stdin: bool=False, json_output: bool=False, color: bool=True, from_pr: str='') -> int | None

#`cmd_doctor`

python

def cmd_doctor(ctx, profile: str='') -> int | None

#`cmd_config`

python

def cmd_config(ctx, profile: str='') -> int | None

#`EventPrinter`

Stateful event printer that deduplicates AssistantText against StreamDelta.

#`print_event`

python

def print_event(self, event: Event) -> None

Pretty-print an event to stdout, deduplicating AssistantText.

#`_print_json`

python

def _print_json(event: Event) -> None

Print an event as a JSON line.

#`main`

python

def main() -> None

The CLI is built with strictcli and provides commands that construct a SessionConfig from flags and run sessions. Commands include send (display response events), stream (real-time token output via StreamDelta), events (raw protocol debug), repl (multi-turn interactive), ask (one-shot text output), doctor (environment health check), config (show resolved config), and the agent subcommand group.

#Event lifecycle

When you call session.send("prompt"), the event flows through an 8-step pipeline that transforms your prompt into a stream of typed events. The steps are: message serialization, subprocess processing, event reading, permission and MCP handling, event flattening and enrichment, file tracking, callback firing, and turn completion with budget checks:

Message serialization. The prompt is wrapped in a UserMessage and written to the subprocess stdin as an NDJSON line via write_message().

Subprocess processing. The Claude Code CLI processes the prompt, makes API calls, and writes events to stdout as NDJSON lines. Events arrive in order: SystemInit (on first turn only), then interleaved AssistantMessage and ToolResultMessage events as the model thinks and uses tools, with possible PermissionRequest and McpRequest control requests, and finally a Result event marking the end of the turn.

Event reading. The session's _read_turn() method reads stdout lines, JSON-decodes them, and calls parse_event() to produce typed events.

Permission and MCP handling. Before yielding, the session checks each event. PermissionRequest events are passed to _handle_permission() which applies the sandbox policy. McpRequest events are routed to _handle_mcp_request() which dispatches tool calls to registered handlers.

Flattening and enrichment. Unless raw=True was passed, events go through flatten_event() to expand compound messages into individual typed events. The session then enriches flattened events: ToolUse events record their tool name by tool_use_id, and later ToolResult events get tool_name stamped from that correlation map.

File tracking. FileWrite and FileEdit events (derived during flattening) accumulate their paths in session.files_modified.

Callback firing. Before yielding each event, the session fires any registered callbacks for that event type.

Turn completion. When a Result event arrives, the session updates cumulative stats (turn count, total tokens, total cost), checks budget thresholds, writes to the cost log if configured, fires on_turn_complete hooks, and returns.

#The streaming model

#Async iteration

The core API is an async generator that yields typed events one at a time as they arrive from the subprocess. Each call to AsyncSession.send() drives one conversational turn, blocking until the subprocess produces each event and terminating when a Result event signals turn completion:

python

async with AsyncSession(config) as session:
    async for event in session.send("prompt"):
        match event:
            case AssistantText(text=t):
                print(t, end="")
            case ToolUse(name=n):
                print(f"[tool: {n}]")
            case Result() as r:
                print(f"\ncost=${r.total_cost_usd:.4f}")

The iterator blocks until the subprocess produces the next event. A turn is complete when a Result event is yielded. Multi-turn conversations call send() multiple times on the same session.

#Raw vs. flattened mode

By default, send() flattens compound events into individual typed events: an AssistantMessage containing 3 content blocks (text, tool_use, thinking) becomes 3 separate events (AssistantText, ToolUse, Thinking). Passing raw=True disables flattening and yields the original compound events with their content block lists intact, which is useful for protocol debugging or custom renderers.

Passing raw=True yields the protocol-level events (AssistantMessage, ToolResultMessage) with their content block lists intact. This is useful for debugging, protocol inspection, or building custom renderers that need the full message structure.

python

# Flattened (default): individual typed events
for event in session.send("prompt"):
    if isinstance(event, AssistantText): ...
    if isinstance(event, ToolUse): ...

# Raw: compound message events with content blocks
for event in session.send("prompt", raw=True):
    if isinstance(event, AssistantMessage):
        for block in event.content:
            if isinstance(block, TextBlock): ...
            if isinstance(block, ToolUseBlock): ...

#Real-time streaming

StreamDelta events carry partial tokens as they arrive from the API. They appear alongside the full AssistantText events (which contain the complete text once the message finishes). The CLI's stream command uses StreamDelta.text for real-time output and falls back to AssistantText if the streamed text differs:

python

streamed = ""
for event in session.send("prompt"):
    if isinstance(event, StreamDelta) and event.text:
        streamed += event.text
        sys.stdout.write(event.text)
    elif isinstance(event, AssistantText):
        if event.text != streamed:
            sys.stdout.write(event.text)

#Event filtering by type

Use isinstance checks or structural pattern matching to filter from the 19 event types in the SDK. The type hierarchy is flat with all events inheriting directly from Event, so filtering requires only single-level dispatch without navigating a deep class tree or handling intermediate abstract types:

python

from claudestream import (
    AssistantText, ToolUse, ToolResult, Thinking,
    Result, PermissionRequest, BudgetThreshold,
)

for event in session.send("prompt"):
    match event:
        case AssistantText():   ...  # Model text output
        case ToolUse():         ...  # Tool call (name + input)
        case ToolResult():      ...  # Tool output
        case Thinking():        ...  # Extended thinking
        case Result():          ...  # Turn complete
        case PermissionRequest(): ...  # Needs permission decision
        case BudgetThreshold(): ...  # Budget threshold crossed

#Callbacks

Register callbacks for specific event types using session.on(EventType, handler). Callbacks fire during iteration before each event is yielded to the consumer, allowing side effects like logging, metrics collection, or progress reporting without modifying the main iteration loop:

python

session.on(ToolUse, lambda e: print(f"[calling {e.name}]"))
session.on(Result, lambda e: print(f"[${e.total_cost_usd:.4f}]"))

#Permission handling

Permission handling has 2 modes: automatic (sandbox-driven, where the SDK resolves tool permission requests using declarative allow/deny rules) and manual (consumer-driven, where PermissionRequest events are surfaced to the caller for interactive decision-making).

#Automatic: sandbox policies

When a Sandbox is configured, the session automatically resolves permission requests by applying a 2-step check (tool allow-list, then write-path scope) without surfacing any events to the consumer. This is the default mode for agent definitions that declare a sandbox.

#`claudestream.policy`

Sandbox and permission policy types for Claude Code sessions, defining allow, deny, and approval rules for tool execution requests.

#`Allow`

Allow the tool to execute.

#`Deny`

Deny the tool execution.

#`Sandbox`

Declarative sandbox configuration for a Claude Code session.

Controls which tools are available, filesystem scope, and behavior flags.

#`create_sandbox`

python

def create_sandbox(*, tools: list[str] | None=None, bare: bool=False, write_paths: list[str] | None=None, log_violations: bool=False, skip_permissions: bool=False) -> Sandbox

Create a validated Sandbox configuration.

Raises:

ValueError: If any tool name is empty or not a string.

#`sandbox_to_flags`

python

def sandbox_to_flags(sandbox: Sandbox | None) -> list[str]

Convert a Sandbox to CLI flags for Claude Code.

None means no sandbox flags (use defaults).

#`_resolve_path`

python

def _resolve_path(path: str, cwd: str) -> str

Resolve a path to an absolute, symlink-free canonical form.

#`_is_within`

python

def _is_within(target: str, allowed: str) -> bool

Check if target is within allowed directory (both must be realpath'd).

Uses string-prefix comparison with a trailing separator to avoid '/src/foo' matching '/src/foobar'.

#`sandbox_decide`

python

def sandbox_decide(sandbox: Sandbox, tool_name: str, tool_input: dict, cwd: str) -> Allow | Deny

Decide whether a tool call is allowed under the given Sandbox.

The Sandbox is the complete authority -- this always returns Allow or Deny, never None.

When a Sandbox is configured, the session automatically resolves permission requests without surfacing them to the consumer. The sandbox_decide() function applies two checks in order:

Tool allow-list. If sandbox.tools is set and the tool name is not in the list, the request is denied.
Write-path scope. If sandbox.write_paths is set and the tool is a write tool (Write, Edit, MultiEdit), the file path is resolved to an absolute canonical path and checked against the allowed directories.

If both checks pass, the request is allowed. The session sends an AllowPermission or DenyPermission message back to the subprocess via stdin.

python

from claudestream import create_sandbox, SessionConfig, SyncSession

# Only allow Read and Bash, restrict writes to one directory
sandbox = create_sandbox(
    tools=["Read", "Bash", "Write"],
    write_paths=["/home/user/project/src"],
)
config = SessionConfig(model="sonnet", profile="default", sandbox=sandbox)

with SyncSession(config) as session:
    result = session.ask("Read the README")
    # Write attempts outside /home/user/project/src are denied automatically

The skip_permissions=True option bypasses all permission prompts by passing --dangerously-skip-permissions to the subprocess. This is for testing only.

#Manual: consumer-driven permission handling

When intercept_permissions=True is set on SessionConfig, or when iterating with raw=True, the session surfaces PermissionRequest events containing the tool name, input parameters, and display metadata. The consumer inspects each request and responds with respond_allow() or respond_deny() to unblock the subprocess:

python

config = SessionConfig(
    model="sonnet",
    profile="default",
    intercept_permissions=True,
)

with SyncSession(config) as session:
    for event in session.send("edit the config file", raw=True):
        if isinstance(event, PermissionRequest):
            # Inspect and decide
            if event.tool_name in ("Read", "Bash"):
                session.respond_allow(event.request_id, event.tool_input)
            else:
                session.respond_deny(event.request_id, "Not allowed")

PermissionRequest events carry rich metadata: tool_name, tool_input, decision_reason, permission_suggestions, and display fields (title, display_name, description) for building UI permission cards.

#User dialog requests

UserDialogRequest events represent blocking dialogs the CLI asks the host to render (e.g., AskUserQuestion). These are never auto-handled by the sandbox. Respond with respond_dialog() to complete the dialog or respond_dialog_cancelled() to let the CLI apply its default behavior.

#Common usage patterns

#One-shot ask

The simplest pattern for single-question interactions. The ask() method internally calls send(), collects all AssistantText events, concatenates their text content, and returns an AskResult containing the full response text along with cost, duration, and token usage metadata:

python

from claudestream import SessionConfig, SyncSession

config = SessionConfig(model="sonnet", profile="default")
with SyncSession(config) as session:
    result = session.ask("What is the capital of France?")
    print(result.text)
    print(f"Cost: ${result.cost_usd:.4f}, Duration: {result.duration_ms:.0f}ms")

#Multi-turn conversation

The subprocess maintains full conversation state across multiple calls to send(), so each subsequent prompt has access to the complete history of prior turns. The session tracks cumulative cost, token usage, and turn count across the entire conversation:

python

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

#Registering custom tools

The @tool decorator creates a Tool from a function's type hints and docstring, automatically generating a JSON Schema for the input parameters. Tools are served to Claude Code via MCP during the session startup handshake, and the SDK dispatches incoming McpRequest events to the registered handler functions:

python

from claudestream import tool, SessionConfig, SyncSession, AssistantText

@tool("my_server")
def search_docs(query: str, max_results: int = 5) -> str:
    """Search the documentation.

    Args:
        query: Search query string.
        max_results: Maximum results to return.
    """
    return f"Found {max_results} results for '{query}'"

config = SessionConfig(
    model="sonnet",
    profile="default",
    tools=[search_docs._tool],
)
with SyncSession(config) as session:
    for event in session.send("Search for authentication docs"):
        if isinstance(event, AssistantText):
            print(event.text, end="")

#Lifecycle hooks

Register hooks for 3 lifecycle events: turn completion (fires after each Result with cost and turn data), errors (fires on unhandled exceptions during iteration), and session close (fires when the context manager exits):

python

def on_done(session, result):
    print(f"Turn {session.turn_count}: {result.num_turns} turns, ${result.total_cost_usd:.4f}")

def on_error(session, exc):
    print(f"Error: {exc}")

config = SessionConfig(model="sonnet", profile="default")
with SyncSession(config) as session:
    session.on_turn_complete(on_done)
    session.on_error(on_error)
    for event in session.send("Do something"):
        pass

#Budget observation

Budget thresholds are informational BudgetThreshold events fired when cumulative cost (USD), turn count, or token count crosses any of the configured threshold values. Each threshold fires exactly once per session, and the event carries the metric name, threshold value, and current value for logging or abort decisions:

python

from claudestream import Budget, SessionConfig, SyncSession, BudgetThreshold

config = SessionConfig(
    model="sonnet",
    profile="default",
    budget=Budget(
        cost_thresholds=[0.01, 0.05, 0.10],
        turn_thresholds=[5, 10],
    ),
)
with SyncSession(config) as session:
    for event in session.send("Do a complex task"):
        if isinstance(event, BudgetThreshold):
            print(f"Budget: {event.metric} crossed {event.threshold} (now {event.current_value})")

#Mid-session control

The session supports 4 mid-session control operations that modify the running subprocess without restarting it: switching the active model, changing permission modes, querying context window usage (total and maximum tokens), and interrupting a running turn to reclaim control:

python

async with AsyncSession(config) as session:
    # Switch model mid-session
    await session.set_model("claude-opus-4-20250514")

    # Query context window usage
    usage = await session.get_context_usage()
    print(f"Context: {usage.total_tokens}/{usage.max_tokens} tokens")

    # Interrupt a running turn
    still_queued = await session.interrupt()

#Agent definitions

Agents are .agent.json files that compose a model, system prompt template, sandbox policy, budget constraints, tool schemas, and MCP configuration into a single reusable definition, loadable by name from .claudestream/agents/ or by filesystem path:

python

from claudestream import load_agent, invoke_agent_sync, SessionConfig, AssistantText

agent = load_agent("code_reviewer")
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="")

#Transparent recovery

When the subprocess becomes unresponsive (no events for stuck_timeout seconds, default 120s), the session automatically restarts it with --resume to preserve conversation state. The recovery sends a random continuation message ("continue", "carry on", etc.) and retries up to 3 times. The restart_count property tracks how many restarts have occurred.

#Architecture Guide

#The four layers

#Process (bottom)

#claudestream._process

#_kill_active_children

#find_binary

#_version_lt

#check_version

#ProcessConfig

#build_argv

#ProcessManager

#stdin

#stdout

#is_alive

#stderr_lines

#_drain_stderr

#start

#close

#kill

#Protocol (middle-lower)

#claudestream._protocol

#parse_content_block

#parse_usage

#parse_event

#_resolve_path

#_derive_file_events

#flatten_event

#read_events

#write_message

#Session (middle-upper)

#claudestream._async_session

#ClaudeStreamError

#AsyncSession

#_enrich_flattened

#_build_process_config

#session_id

#model_name

#tools

#claude_version

#last_result

#turn_count

#total_tokens

#total_cost_usd

#stderr_lines

#sandbox

#user_tools

#is_alive

#active_turn

#cancelled

#files_modified

#process_pid

#cwd

#mcp_servers

#permission_mode

#restart_count

#config

#_start

#_read_control_response

#_run_mcp_handshake

#close

#cancel

#send

#ask

#_read_turn

#_liveness_probe

#_restart_subprocess

#_handle_permission

#_handle_mcp_request

#_resolve_control

#_fail_pending_controls

#_control_request

#_read_control_result

#_check_thresholds

#_write_cost_log

#on

#on_turn_complete

#on_error

#on_close

#_fire_hooks

#respond_allow

#`claudestream._process`

#`_kill_active_children`

#`find_binary`

#`_version_lt`

#`check_version`

#`ProcessConfig`

#`build_argv`

#`ProcessManager`

#`stdin`

#`stdout`

#`is_alive`

#`stderr_lines`

#`_drain_stderr`

#`start`

#`close`

#`kill`

#`claudestream._protocol`

#`parse_content_block`

#`parse_usage`

#`parse_event`

#`_resolve_path`

#`_derive_file_events`

#`flatten_event`

#`read_events`

#`write_message`

#`claudestream._async_session`

#`ClaudeStreamError`

#`AsyncSession`

#`_enrich_flattened`

#`_build_process_config`

#`session_id`

#`model_name`

#`tools`

#`claude_version`

#`last_result`

#`turn_count`

#`total_tokens`

#`total_cost_usd`

#`stderr_lines`

#`sandbox`

#`user_tools`

#`is_alive`

#`active_turn`

#`cancelled`

#`files_modified`

#`process_pid`

#`cwd`

#`mcp_servers`

#`permission_mode`

#`restart_count`

#`config`

#`_start`

#`_read_control_response`

#`_run_mcp_handshake`

#`close`

#`cancel`

#`send`

#`ask`

#`_read_turn`

#`_liveness_probe`

#`_restart_subprocess`

#`_handle_permission`

#`_handle_mcp_request`

#`_resolve_control`

#`_fail_pending_controls`

#`_control_request`

#`_read_control_result`

#`_check_thresholds`

#`_write_cost_log`

#`on`

#`on_turn_complete`

#`on_error`

#`on_close`

#`_fire_hooks`

#`respond_allow`

#`respond_deny`

#`respond_dialog`

#`respond_dialog_cancelled`

#`interrupt`

#`set_permission_mode`