claudestream v0.14.2 /claudestream._sync_session
Updated
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.

Search