claudestream v0.14.2 /claudestream._async_session
Updated
On this page

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

#claudestream._async_session

#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:

  1. Send InitializeRequest -> read ControlResponse
  2. Send McpSetServers -> read ControlResponse
  3. 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

Register a callback for a specific event type.

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

#on_turn_complete

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

Register a hook that fires after each turn completes (after Result event).

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

Register a hook that fires when a turn fails with an exception.

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

Register a hook that fires when the session closes.

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.

Search