Transports

A WireStream is A11's pluggable transport — a bidirectional, unordered channel between two endpoints. Choosing an implementation is how an agent goes from in-process to networked; see Principles.

WireStream

start and accept install callbacks on the initiating and responding sides. send admits a message into the bounded outgoing path; half_close queues the local end marker; and drain_outgoing_messages is the delivery barrier. abort terminates a failed exchange with a structured status.

a11.net.wire_stream.WireStream

WireStream()

Construct the abstract WireStream base. Subclass this in Python to implement a custom asynchronous, bidirectional transport for an agent; the abstract operations (send, start/accept, get_status, get_trailers, ...) are dispatched to your overrides.

deadline property

deadline: Any

The stream's current absolute deadline, after which it is automatically aborted.

abort

abort(status: Any) -> None

Terminate the stream immediately with an error status, discarding buffered messages and propagating failure to the peer and pending receivers.

Examples:

End an exchange when its upstream disappears:

stream.abort(Status(
    code=StatusCode.UNAVAILABLE,
    message="upstream connection was lost",
))

accept

accept(on_message: Any, on_done: Any) -> Any

Begin driving the stream as the responding (server) side, delivering each inbound message to the asynchronous on_message callback and end-of-stream to on_done. Use this instead of start() when this endpoint is answering an incoming agent connection. Returns an awaitable that resolves when acceptance completes; use on_done as the terminal barrier.

drain_outgoing_messages

drain_outgoing_messages() -> Any

Await until every queued outbound message has been handed to the transport. Call half_close first so buffered output is not dropped.

Examples:

Use the transport delivery barrier during orderly shutdown:

stream.half_close()
await stream.drain_outgoing_messages()

get_id

get_id() -> str

Return the stream's stable identifier, which also seeds its tracing trace id. Use it to correlate an agent stream with logs and traces.

get_impl

get_impl() -> Any

Return an opaque native handle to the underlying implementation, or None. Intended for advanced interop, not normal agent code.

get_status

get_status() -> Any

Return the stream's terminal status once it has finished, or OK while it is still active. Inspect this after the stream completes to learn whether the agent exchange succeeded or failed.

get_trailers

get_trailers() -> Any

Return the trailers (final metadata) the peer sent at half-close, or None if none were received. Read this after the stream ends to recover end-of-turn metadata from the agent exchange.

half_close

half_close(trailers: Any | None = None) -> None

Signal that this endpoint has finished sending, optionally attaching trailers. The stream stays open for inbound messages.

Examples:

End the local half and wait until queued messages reach the transport:

stream.half_close()
await stream.drain_outgoing_messages()

send

send(message: WireMessage) -> None

Queue a message for asynchronous delivery to the peer. This call is non-blocking: the message enters the ordered outbound queue and the transport applies backpressure.

Examples:

Admit a request before closing the local sending side:

stream.send(request_message)

set_deadline

set_deadline(deadline: Any | None = None) -> None

Set an absolute wall-clock deadline after which the stream is automatically aborted; pass None to clear it. Use this to bound how long an agent interaction is allowed to run.

start

start(on_message: Any, on_done: Any) -> Any

Begin driving the stream as the initiating side, delivering inbound messages to on_message and completion to on_done. Callbacks are awaited as data arrives.

Examples:

Start a client transport with application callbacks:

await stream.start(on_message, on_transport_done)

a11.net.wire_stream.WireStreamOptions

WireStreamOptions(max_buffered_incoming_messages: Any = 100, max_single_message_size: Any = 33554432, max_buffered_incoming_bytes: Any = 33554432, message_timeout_millis: Any | None = None, deadline: Any | None = None)

Construct wire-stream options controlling buffering and timeouts for an agent stream. All arguments are keyword-friendly and validated on construction.

deadline property writable

deadline: Any

Absolute wall-clock deadline after which the stream is aborted.

max_buffered_incoming_bytes property writable

max_buffered_incoming_bytes: int

Maximum total bytes of buffered inbound messages before backpressure is applied.

max_buffered_incoming_messages property writable

max_buffered_incoming_messages: int

Maximum number of inbound messages buffered before backpressure is applied.

max_single_message_size property writable

max_single_message_size: int

Maximum size, in bytes, of a single wire message.

message_timeout property writable

message_timeout: Any

Per-message inactivity timeout as a duration.

message_timeout_millis property writable

message_timeout_millis: Any

Per-message inactivity timeout expressed in milliseconds.

validate

validate() -> None

Validate the options, raising on invalid configuration.

In-process

a11.net.in_process_wire_stream.InProcessWireStream

InProcessWireStream()

Bases: WireStream

create_pair staticmethod

create_pair(options: WireStreamOptions | None = None, first_options: WireStreamOptions | None = None, second_options: WireStreamOptions | None = None) -> tuple[InProcessWireStream, InProcessWireStream]

Create a connected pair of in-process wire streams that talk to each other directly in memory, with no network involved. Use this to wire an agent to a local service or test harness: one endpoint drives start() while the other drives accept(). Pass shared options, or per-endpoint first_options/second_options, to tune buffering and timeouts.

wait

wait() -> Any

Await until this in-process stream has fully finished. Block on this to know a local agent exchange has completed before tearing the pair down.

a11.net.in_process_wire_stream.create_in_process_wire_stream_pair

create_in_process_wire_stream_pair(options: WireStreamOptions | None = None, *, first_options: WireStreamOptions | None = None, second_options: WireStreamOptions | None = None) -> tuple[InProcessWireStream, InProcessWireStream]

Use create_in_process_wire_stream_pair to test both endpoints without opening a socket.

WebSocket

a11.net.websocket_wire_stream.WebSocketWireStream

WebSocketWireStream()

Bases: WireStream

connect staticmethod

connect(url: str, options: WireStreamOptions = ..., websocket_options: WebSocketClientOptions = ...) -> WebSocketWireStream

Open a client WebSocket connection to url and return a WireStream over it. This is the standard way for an agent to dial out to a remote A11 endpoint; the returned stream is then driven asynchronously via start()/send(). Tune transport buffering with options and the handshake (headers, framing, HTTP/2, TLS) with websocket_options.

a11.net.websocket_wire_stream.WebSocketWireServer

port property

port: int

The actual TCP port the server is listening on, resolved even when an ephemeral port (0) was requested.

running property

running: bool

Whether the server is currently accepting connections.

create staticmethod

create(on_stream: Any, options: WebSocketServerOptions = ...) -> WebSocketWireServer

Start a WebSocket server that accepts incoming A11 connections, invoking the asynchronous on_stream callback with a fresh WireStream for each accepted client. This is the server-side entry point for hosting an agent: each callback runs concurrently and typically drives accept() on its stream. Configure the listen address, port, path and TLS via options.

get_impl

get_impl() -> Any

Return an opaque native handle to the underlying implementation, or None. Intended for advanced interop.

stop

stop() -> None

Stop the server and close the listening socket, releasing the bound port. Call this to shut the agent host down cleanly; it blocks until shutdown completes.

HTTP SSE

a11.net.http_sse_wire_stream.HttpSseWireStream

HttpSseWireStream()

Bases: WireStream

get_http_request_headers

get_http_request_headers() -> list

Return the HTTP headers carried on the underlying SSE request. This is the base class shared by the client and server SSE wire streams that transport A11 messages over an HTTP/2 Server-Sent Events connection. Use it when building an agent that needs to inspect the transport-level request metadata.

get_http_response_headers

get_http_response_headers() -> Any

Return the HTTP response headers negotiated for the SSE connection, or None if they have not arrived yet. Because the connection is established asynchronously, prefer awaiting wait_for_http_headers() before relying on this value.

set_http_request_headers

set_http_request_headers(headers: Any) -> None

Set the HTTP headers to send on the underlying SSE request. Call this before the stream connects to attach auth or routing metadata that your agent's transport needs.

set_http_response_headers

set_http_response_headers(headers: Any) -> None

Set the HTTP headers to send on the SSE response. Used on the server side to attach transport metadata before the streaming response is flushed to the client.

wait_for_http_headers

wait_for_http_headers() -> Any

Await the exchange of HTTP headers for the SSE connection. Because SSE wire streams connect asynchronously, await this future before reading response headers or assuming the stream is live.

a11.net.http_sse_wire_stream.HttpSseServer

http2_server property

http2_server: Http2Server

The underlying HTTP/2 server.

port property

port: int

The port the server is listening on.

running property

running: bool

Whether the server is currently running.

create staticmethod

create(bind_address: str = '127.0.0.1', port: SupportsInt | SupportsIndex = 0, on_connect: Any | None = None, options: HttpSseOptions = ...) -> HttpSseServer

Create and start an SSE server that accepts A11 wire streams, invoking the optional async on_connect callback for each client.

stop

stop() -> None

Stop the server and release its resources.

wait_for_stream

wait_for_stream() -> Any

Await the next incoming SSE wire stream from a connecting client.

WebRTC

a11.net.webrtc_wire_stream.WebRtcWireStream

WebRtcWireStream()

Bases: WireStream

data_channel property

data_channel: Any

Opaque capsule around the underlying libdatachannel DataChannel. Exposed for advanced interop and diagnostics; agent code normally reads and writes through the WireStream API rather than touching this directly.

peer_connection property

peer_connection: Any

Opaque capsule around the underlying libdatachannel PeerConnection. Useful for inspecting ICE/connection state during debugging; not required for normal streaming.

signalling_endpoint property

signalling_endpoint: SignallingTransport

Signalling transport this stream negotiated over. Lets an agent observe or reuse the channel that carried the asynchronous SDP/ICE handshake.

a11.net.webrtc_wire_stream.WebRtcWireServer

identity property

identity: str

Local identity this server listens as.

pending_peer_count property

pending_peer_count: int

Number of peers still completing negotiation.

running property

running: bool

Whether the server is currently running.

signalling_endpoint property

signalling_endpoint: SignallingEndpoint

Signalling endpoint the server negotiates over.

create staticmethod

create(identity: str, signalling: SignallingService, on_stream: Any, configuration: WebRtcConfiguration = ..., stream_options: WireStreamOptions = ...) -> WebRtcWireServer

Create a WebRTC server that accepts peer connections and invokes the async on_stream callback with each new WebRtcWireStream.

stop

stop() -> None

Stop the server and stop accepting new peer connections.

Signalling

Signalling is the out-of-band handshake WebRTC peers use to find each other and exchange connection details.

For a service, WebSocketSignallingServer.create starts the endpoint and stop ends acceptance during shutdown.

a11.net.signalling.WebSocketSignallingServer

port property

port: int

Port the server is listening on.

running property

running: bool

Whether the server is currently running.

service property

service: SignallingService

Signalling service this server fronts.

create staticmethod

create(service: SignallingService, options: WebSocketSignallingServerOptions = ...) -> WebSocketSignallingServer

Create a WebSocket signalling server that fronts the given in-process signalling service.

get_impl

get_impl() -> Any

Opaque capsule around the native implementation, for interop.

stop

stop() -> None

Stop the server and close all client connections.

a11.net.signalling.WebSocketSignallingClient

WebSocketSignallingClient()

Bases: SignallingTransport

connect staticmethod

connect(url: str, identity: str, on_message: Any | None = None, options: WebSocketSignallingClientOptions = ...) -> Any

Asynchronously connect to a WebSocket signalling server, resolving to a client once registered under the given identity.

get_impl

get_impl() -> Any

Opaque capsule around the native implementation, for interop.

a11.net.signalling.SignallingService

SignallingService()

Create a new in-process signalling service.

create staticmethod

create() -> SignallingService

Create a new in-process signalling service.

connect

connect(identity: str, on_message: Any) -> SignallingEndpoint

Register an identity and its async inbound-message callback, returning a signalling endpoint.

contains

contains(identity: str) -> bool

Return whether the given identity is currently connected.

identities

identities() -> list[str]

Return the list of currently connected identities.

stop

stop() -> None

Stop the service and disconnect all endpoints.

HTTP/2 primitives

Low-level building blocks under the WebSocket and SSE transports; most agents use them only indirectly.

a11.net.http2.Http2Client

connected property

connected: bool

Whether the client is currently connected.

host property

host: str

The host the client is connected to.

port property

port: int

The port the client is connected to.

secure property

secure: bool

Whether the connection is using TLS.

connect staticmethod

connect(host: str, port: SupportsInt | SupportsIndex, options: Http2Options = ...) -> Any

Asynchronously connect to an HTTP/2 server, returning a future that resolves to the connected client.

close

close() -> None

Close the client connection.

extended_connect

extended_connect(protocol: str, path: str, headers: Any | None = None, scheme: str = '') -> Http2DuplexStream

Open an extended CONNECT duplex stream for bidirectional data.

get_impl

get_impl() -> Any

Return an opaque capsule wrapping the native client handle.

request

request(method: str, path: str, headers: Any | None = None, body: Any = b'', scheme: str = '') -> Any

Send a request and await the full buffered response.

request_stream

request_stream(method: str, path: str, headers: Any | None = None, body: Any = b'', scheme: str = '') -> Http2ResponseStream

Open a request and return a pull-oriented response stream for reading the response body incrementally.

a11.net.http2.Http2Server

bind_address property

bind_address: str

The address the server is bound to.

port property

port: int

The port the server is listening on.

running property

running: bool

Whether the server is currently running.

secure property

secure: bool

Whether the server is using TLS.

create staticmethod

create(bind_address: str = '127.0.0.1', port: SupportsInt | SupportsIndex = 0, handler: Any | None = None, options: Http2Options = ...) -> Http2Server

Create and start an HTTP/2 server bound to the given address and port, dispatching each request to the async handler.

get_impl

get_impl() -> Any

Return an opaque capsule wrapping the native server handle.

stop

stop() -> None

Stop the server and release its resources.