Sessions

A Session is the connection-scoped runtime: it multiplexes wire streams, dispatches inbound action calls, and manages their lifetimes. It is the object you build a server or client agent around.

Session

Attach a transport with add_stream and route outbound messages with send. half_close begins an orderly shutdown; await done when every stream and dispatched action must have released its state. Use abort for a failed connection so the peer receives a structured reason.

a11.service.session.Session

Session(session_id: str = '', on_stream_message: Any | None = None, on_stream_done: Any | None = None, headers: Any | None = None, options: SessionOptions | None = None, node_map: NodeMap | None = None, action_registry: ActionRegistry | None = None)

Create an A11 session that multiplexes wire streams and actions. Streams deliver messages asynchronously to the optional on_stream_message and on_stream_done callbacks, which may be coroutines. This is the top-level object an agent drives to exchange wire messages and run actions.

action_registry property writable

action_registry: ActionRegistry | None

The ActionRegistry used to resolve action messages; assigning replaces it.

deadline property

deadline: Any

The absolute time after which the session will be aborted.

done property

done: _DoneEvent

An asyncio.Event-shaped view of full session completion.

Session.is_closed can become true as soon as shutdown starts. Await this event (or wait_done) when streams and actions must all have released their runtime state.

id property

id: str

The session's unique identifier string.

node_map property writable

node_map: NodeMap

The NodeMap backing this session's node state; assigning replaces it.

abort

abort(status: Any) -> None

Abort the session immediately with the given error status, cancelling streams and actions. Use this when an unrecoverable error occurs.

Examples:

Propagate an authentication failure to the peer:

session.abort(Status(
    code=StatusCode.PERMISSION_DENIED,
    message=str(error),
))

actions

actions() -> list[tuple[str, Action]]

Return the (action_id, action) pairs currently running in the session. Actions execute asynchronously, so this is a point-in-time snapshot of in-flight work.

add_stream

add_stream(stream: WireStream, mode: Any = 'start') -> Any

Attach a wire stream and begin pumping its messages, returning an awaitable for the stream's lifetime. mode selects whether this side starts ("start") or accepts ("accept") the stream.

Examples:

Attach the client transport before exchanging messages:

stream_lifetime = session.add_stream(websocket_stream)

await_all_actions

await_all_actions(timeout: Any | None = None) -> Any

Return an awaitable that resolves once all in-flight actions have finished, or the optional timeout elapses. Await this to synchronize on the session's outstanding asynchronous work before proceeding.

cancel_action

cancel_action(action_id: str) -> None

Request cancellation of the running action with the given id, raising if it is unknown. Cancellation is cooperative and completes asynchronously as the action unwinds.

cancel_all_actions

cancel_all_actions() -> None

Request cancellation of every action currently running in the session. Each action unwinds asynchronously; await await_all_actions to observe completion.

dispatch_action

dispatch_action(action: Any) -> Any

Dispatch an already-constructed Action to run within the session, returning an awaitable for its handling. Use this to inject actions programmatically rather than via an incoming wire message.

dispatch_action_message

dispatch_action_message(action_message: ActionMessage, origin_stream: WireStream | None = None) -> Any

Dispatch an action message, resolving it against the action registry and running the resulting action. Returns an awaitable that completes when the action has been handled; origin_stream attributes the message to a source stream.

dispatch_node_fragment

dispatch_node_fragment(fragment: NodeFragment) -> Any

Dispatch a node fragment into the session's NodeMap and return an awaitable resolving to the applied revision. Fragments are applied asynchronously in order, letting an agent stream incremental document updates.

dispatch_wire_message

dispatch_wire_message(message: WireMessage, origin_stream: WireStream | None = None) -> Any

Route a wire message through the session as though it arrived on a stream, returning an awaitable for its processing. origin_stream optionally records which stream the message is attributed to.

get_action

get_action(action_id: str) -> Action

Look up a running action by its id, raising if none matches. Useful for inspecting or awaiting a specific asynchronous action you previously dispatched.

get_action_registry

get_action_registry() -> ActionRegistry | None

Return the ActionRegistry used to resolve incoming action messages into runnable actions.

get_id

get_id() -> str

Return the session's unique identifier string. Use it to correlate this session with logs, traces, and external bookkeeping while it runs asynchronously.

get_node_map

get_node_map() -> NodeMap

Return the NodeMap backing this session's node state. Node fragments dispatched to the session are applied to this map as messages stream in.

get_status

get_status() -> Any

Return the session's terminal status, indicating whether it completed successfully or was aborted.

get_stream

get_stream(stream_id: str) -> WireStream

Look up an attached stream by its id, raising if no such stream exists. Because streams come and go over the session's lifetime, guard against a stream having been removed since you last observed it.

half_close

half_close() -> None

Signal that this side will send no more messages, allowing the session to drain and finish once peers do the same. Remaining inbound messages continue to be processed asynchronously.

Examples:

Finish an exchange after sending the last message:

session.half_close()
await session.done.wait()

is_closed

is_closed() -> bool

Return whether the session has been closed and no longer accepts new streams or messages.

is_done

is_done() -> bool

Return whether the session has fully finished, including all streams and actions. Prefer awaiting done for asynchronous completion rather than polling this flag.

send

send(message: WireMessage, stream_id: str = '') -> None

Enqueue a wire message for delivery on the named stream (or the default stream), raising on failure. Delivery happens asynchronously as the stream drains.

Examples:

Route a response through a particular attached transport:

session.send(response, stream_id=websocket_stream.get_id())

set_action_registry

set_action_registry(registry: ActionRegistry | None) -> None

Replace the ActionRegistry used to resolve incoming action messages, raising on failure. Active actions are rebound for later nested-name resolution; configure it before dispatch to avoid mixing registry versions.

set_deadline

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

Set the absolute deadline after which the session is aborted; passing None clears it to no deadline. The session enforces this asynchronously as time passes.

set_node_map

set_node_map(node_map: NodeMap) -> None

Replace the NodeMap backing this session's node state, raising on failure. Active actions are rebound, but existing fragments are not migrated; set it before traffic to avoid splitting state between maps.

streams

streams() -> list[tuple[str, WireStream]]

Return the (stream_id, stream) pairs currently attached to the session. Streams are added and removed asynchronously as peers connect and disconnect, so treat the result as a snapshot taken at call time.

wait_done

wait_done() -> Any

Return an awaitable that resolves when the session has fully finished. Await this to block until every stream and action has completed asynchronously.

SessionWithRecv

receive is the pull-style flow for one transport, while receive_with_stream_id retains the source when a gateway multiplexes clients.

a11.service.session.SessionWithRecv

SessionWithRecv(session_id: str = '', headers: Any | None = None, options: SessionOptions | None = None, node_map: NodeMap | None = None, action_registry: ActionRegistry | None = None)

Bases: Session

Create a session that buffers inbound messages for explicit pull-based reception instead of callbacks. Use receive or receive_with_stream_id to await messages as they stream in, which suits agents that consume messages in their own loop.

receive async

receive(deadline=None)

Await the next inbound message, or None when the session ends.

Use this when one receive loop handles every attached stream. Choose receive_with_stream_id when replies or diagnostics must retain their transport identity. The optional absolute deadline limits only this wait; it does not change the session deadline.

Examples:

Route messages from a session with one attached transport:

while message := await session.receive():
    await route_message(message)

receive_with_stream_id async

receive_with_stream_id(deadline=None)

Await (message, stream_id), or None after completion.

This is the pull-style counterpart to OnSessionStreamMessage and is useful when an agent multiplexes several transports in one loop.

Examples:

Preserve the source while routing gateway traffic:

while item := await session.receive_with_stream_id():
    message, stream_id = item
    await route_message(message, source=stream_id)

SessionOptions

a11.service.session.SessionOptions

SessionOptions(*, max_buffered_messages_total: Any = 256, max_buffered_messages_per_stream: Any = 32, max_concurrent_root_actions: Any = 32, max_concurrent_nested_actions: Any = 128, max_single_message_size: Any = 33554432, max_buffered_bytes_total: Any = 33554432, max_buffered_bytes_per_stream: Any = 4194304, no_stream_timeout: Any | None = None, deadline: Any | None = None)

Construct session limits and timeouts; all parameters are keyword-only.

deadline property writable

deadline: Any

Absolute time after which the session is aborted.

max_buffered_bytes_per_stream property writable

max_buffered_bytes_per_stream: int

Maximum bytes buffered per stream.

max_buffered_bytes_total property writable

max_buffered_bytes_total: int

Maximum total bytes buffered across all streams.

max_buffered_messages_per_stream property writable

max_buffered_messages_per_stream: int

Maximum number of messages buffered per stream.

max_buffered_messages_total property writable

max_buffered_messages_total: int

Maximum number of messages buffered across all streams.

max_concurrent_nested_actions property writable

max_concurrent_nested_actions: int

Maximum number of concurrently running nested actions.

max_concurrent_root_actions property writable

max_concurrent_root_actions: int

Maximum number of concurrently running root actions.

max_single_message_size property writable

max_single_message_size: int

Maximum size in bytes of a single wire message.

no_stream_timeout property writable

no_stream_timeout: Any

How long the session waits with no active stream before finishing.

validate

validate() -> None

Validate the option values, raising on invalid configuration.