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

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