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

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

#claudestream._protocol

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

Search