Skip to content
claudestream._sync_session
Updated Edit
On this page

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

#claudestream._sync_session

#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

Register a callback for a specific event type.

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

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

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

Register a hook that fires when the session closes.

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.

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