Skip to content
claudestream._async_session
Updated Edit
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.

More tools from this site

  • claudewheel A TUI Claude Code Launcher that lets you have more than one profile, manage sessions lifecycle, pick the exact CC version, model to use (even older unlisted ones), pick which GitHub account to use, etc.
  • dirstat Fast, single-binary directory statistics CLI: every file under a tree grouped by format, with counts, sizes, and lines of code, as a colored terminal table or as JSON
  • fastware A batteries-included ASGI framework: msgspec JSON, a managed Granian server, dependency injection, SSE, WebSockets, auth, and a test client
  • go-toml-edit Zero-dep TOML editing library for Go with comment preservation
  • howmuchleft The fastest Claude Code statusline: context window, 5-hour, and weekly limit usage as three customizable gradient bars, rendering in about 6 ms
  • orxtra
  • pgdesign
  • predraw Declarative rendering pipeline: describe a scene in JSON and get SVG, PNG and WebP out, with light and dark style tokens, reusable components and text converted to path outlines
  • reposummary Turn a git repository's history into a Markdown journal: pick a time window or revision range and get a readable digest of what changed, optionally narrated by an LLM
  • rlsbl Release orchestration and project scaffolding CLI that bumps versions, validates a structured JSONL changelog, tags only the commit CI verified, and publishes to npm, PyPI, Go and more
  • safegit git wrapper CLI that gives each commit its own temporary index and retries ref updates on conflict, so concurrent agents share one repository
  • saferm Command-line replacement for rm that archives every deletion with a mandatory reason and the context it ran in, so deleted files can be listed, inspected and restored
  • selfdoc Static Site Generator that builds a project's documentation site directly from its source code, so the docs can never drift from the code they describe, with SEO/AEO, first-class blog, search, and cross-project linking built in
  • strictcli
  • stricttest An always-on test-isolation floor: a pytest plugin and a Go env-hygiene module that make a test suite structurally unable to reach real credentials, the real HOME, the network, or the development repository.
  • wesktop A Python framework that turns an ASGI web app into a desktop application, serving it from a local Granian server and displaying it in a native OS window via pywebview
Search