stdiobus v2.1.1

The SDK provides typed JSON-RPC messaging, session routing, streaming aggregation, and cross-platform backend support for AI agent communication.

Python SDK for building AI agents over stdio_bus — a reliable transport layer for ACP/MCP-style workflows.

Highlights

• Streaming support — agent_message_chunk notifications are aggregated into result["text"] automatically
• Cancellation semantics — stop() fails all pending requests with TransportError, no silent drops
• Notification handler — on_notification() for subscribing to notifications separately from raw messages
• Cross-SDK contract tests — wire-format parity verification with Rust and Node SDKs
• CI pipeline — GitHub Actions for unit, E2E, real-binary, and Docker tests

Since 1.0.0

Key evolution across 1.0.0 → 2.0.0 → 2.1.1:

• SubprocessBackend added — spawns stdio_bus binary via stdin/stdout NDJSON pipes, now the default backend
• Hello handshake — stdio_bus/hello protocol negotiation with hello() and connect()
• Protocol extensions — _ext.identity, _ext.audit, agentId routing in JSON-RPC messages
• Programmatic config — BusConfig / PoolConfig / LimitsConfig dataclasses, no config files needed
• Graceful shutdown — close stdin → drain timeout → SIGTERM → SIGKILL with orphan process cleanup

Full history: CHANGELOG.md

What's Changed in 2.1.1

Added

• Streaming support — agent_message_chunk notification aggregation into result.text (ACP protocol parity with Rust SDK)
• Cancellation semantics — stop() fails all pending requests with TransportError("Bus is shutting down")
• on_notification() handler — subscribe to notifications separately from raw messages
• Cross-SDK contract tests — wire-format parity verification (sessionId, agentId, _ext, config schema, streaming, cancellation)
• CI pipeline — GitHub Actions workflow for unit, E2E, real-binary, and Docker tests
• _PendingRequest internal class — tracks pending requests with streaming chunk aggregation

Changed

• _handle_message now dispatches notifications separately and aggregates streaming chunks
• stop() now cancels all in-flight requests before stopping backend
• Backend crash (_on_backend_closed) uses _PendingRequest wrapper

Compatibility

Requirement	Value
Python	3.10, 3.11, 3.12, 3.13, 3.14
OS	Linux, macOS, Windows
Runtime	Native cffi embed, stdio_bus binary in PATH, or Docker
Dependencies	Zero (stdlib only). Optional: cffi for native backend
License	Apache-2.0

Backends

Backend	When used	Config delivery
subprocess	stdio_bus binary in PATH (default)	--config-fd pipe
native	libstdio_bus.a built with cffi	embed API (in-process)
docker	Docker available, no binary	mounted config file

Auto-selection order: subprocess → native → docker (Unix), subprocess → docker (Windows).

Install

pip install stdiobus

With native backend support:

pip install stdiobus[native]

Quick Start

import asyncio
from stdiobus import AsyncStdioBus, BusConfig, PoolConfig

async def main():
    async with AsyncStdioBus(
        config=BusConfig(
            pools=[PoolConfig(id="echo", command="python", args=["./echo_worker.py"], instances=1)]
        )
    ) as bus:
        result = await bus.request("echo", {"message": "hello"})
        print(result)

asyncio.run(main())

Known Limitations

• No automatic reconnect. If the bus process exits, pending requests fail with TransportError. Create a new instance to reconnect.
• stdout from the bus process must carry NDJSON protocol messages only.

Links

• PyPI: pypi.org/project/stdiobus
• Documentation: README
• Issues: github.com/stdiobus/stdiobus-python/issues
• Changelog: CHANGELOG.md
• Website: stdiobus.com