Nodes

Nodes are A11's ordered, asynchronous streams — the way data moves between action ports and across the network. See Principles for the model.

AsyncNode

Create a stream with create, produce it with put and put_final, then finish the writer with drain_and_close. A consumer can follow the stream with next or async iteration, while consume validates the common case of one complete logical value. Configure repeated reads once with set_expected_types.

a11.nodes.async_node.AsyncNode

AsyncNode(chunk_store: ChunkStore, node_map: NodeMap | None = None, *, serialization_registry: SerializationRegistry | None = None, reader_options: ChunkStoreReaderOptions | dict[str, Any] | None = None, writer_options: ChunkStoreWriterOptions | dict[str, Any] | None = None)

Build a node over chunk_store.

Prefer create, which constructs the store for you from a node id. Pass reader_options / writer_options (as ChunkStoreReaderOptions / ChunkStoreWriterOptions or plain dicts) to tune buffering and ordering, and a custom serialization_registry to control how Python objects map to chunks.

chunk_store property

chunk_store: ChunkStore

The underlying chunk store backing this node (see get_chunk_store).

id property

id: str

The node's stable identifier (see get_id).

reader property

reader: ChunkStoreReader

The node's ChunkStoreReader.

reader_options property writable

reader_options: ChunkStoreReaderOptions

Options controlling how this node reads from its chunk store, such as buffering and flow control.

serialization_registry property writable

serialization_registry: SerializationRegistry

The registry used to (de)serialize Python objects for this node.

writer property

writer: ChunkStoreWriter

The node's ChunkStoreWriter.

writer_options property writable

writer_options: ChunkStoreWriterOptions

Options controlling how this node writes to its chunk store, such as buffering and flow control.

expect_types

expect_types(**kwds) -> Iterator[AsyncNode]

Temporarily set the expected read types for the with block.

create classmethod

create(node_id: str, node_map: NodeMap | None = None, *, serialization_registry: SerializationRegistry | None = None, reader_options: ChunkStoreReaderOptions | dict[str, Any] | None = None, writer_options: ChunkStoreWriterOptions | dict[str, Any] | None = None, chunk_store_factory: Callable[[str], ChunkStore] = LocalChunkStore) -> AsyncNode

Create a standalone node identified by node_id.

chunk_store_factory builds the backing store from the id; it defaults to an in-memory LocalChunkStore, so overriding it is how you place a node's data in a different backend.

Examples:

Create a stream used to deliver answer fragments:

answer = AsyncNode.create("answer-tokens")

abort_with_status

abort_with_status(status: Any) -> Any

Aborts the stream with the given error status and returns a future that resolves once the abort has propagated. Use this to fail a stream so consumers observe the error instead of a normal end-of-stream.

attach_stream

attach_stream(stream: WireStream) -> None

Attaches a wire stream so this node's chunks are mirrored over the network transport. Use this to bridge a local streaming node to a remote peer; the stream is kept alive for the node's lifetime.

cancel

cancel() -> None

Cancels both the reader and the writer, tearing down all pending streaming operations on the node at once.

cancel_reader

cancel_reader() -> None

Cancels the node's reader, unblocking any pending next-chunk or next-fragment awaits on the consuming side of the stream.

cancel_writer

cancel_writer() -> None

Cancels the node's writer, unblocking any pending put or drain awaits on the producing side of the stream.

consume async

consume(obj_type: type[T] | None = None, timeout: Duration | None = None, mimetype_patterns: str | Sequence[str] = '', allow_none: bool = False) -> T | Any | None

Consume exactly one whole value and return it deserialized.

Use this for a node that carries a single result (the common case for a unary action output). Pass obj_type to deserialize to a specific type, or request NodeFragment/Chunk to get the raw form.

Examples:

Read the unary customer input of an action handler:

customer = await action["customer"].consume(obj_type=Customer)

consume_chunk async

consume_chunk(timeout: Duration | None = None, allow_none: bool = False) -> Chunk | None

Consume exactly one whole value and return its raw chunk.

consume_fragment async

consume_fragment(timeout: Duration | None = None, allow_none: bool = False) -> NodeFragment | None

Read exactly one whole value's fragment, enforcing the terminator.

Unlike next_fragment, this expects the node to hold a single (possibly multi-part) value followed by a null final chunk, and raises if that shape is violated. With allow_none an empty stream yields None instead of raising. Requires an ordered reader.

detach_stream

detach_stream(stream: WireStream) -> None

Detaches a previously attached wire stream so the node stops mirroring its chunks over that transport.

drain_and_close

drain_and_close() -> Any

Returns a future that resolves once all buffered chunks have been flushed and the writer is closed. This does not mark a chunk as final: call put_final() or put_null_final() first when readers must synchronise on the logical end of the stream.

get_chunk_store

get_chunk_store() -> ChunkStore

Returns the underlying chunk store backing this node. The chunk store is the ordered storage boundary that the node's reader and writer stream through; reach for it when you need lower-level access than the async put/next API provides.

get_id

get_id() -> str

Returns the node's stable identifier. Use this to correlate a streaming node with the rest of an agent's state, for logging, or to key it in a NodeMap. Raises if the id cannot be resolved.

get_reader_options

get_reader_options() -> ChunkStoreReaderOptions

Return a copy of the reader's current options.

get_reader_status

get_reader_status() -> Any

Returns the current status of the node's reader. Check it to tell whether the consuming end of the stream is healthy, has completed, or has failed while streaming.

get_writer_abort_status

get_writer_abort_status() -> Any

Returns the status the writer was aborted with, or None if the writer has not been aborted. Use this to surface the reason a stream was cut short to the rest of an agent.

get_writer_options

get_writer_options() -> ChunkStoreWriterOptions

Return a copy of the writer's current options.

get_writer_status

get_writer_status() -> Any

Returns the current status of the node's writer. Check it to tell whether the producing end of the stream is healthy, has completed, or has failed while streaming.

is_writable

is_writable() -> Any

Returns a future that resolves once it is known whether the node can currently accept writes. Await it before producing chunks to respect backpressure rather than blocking a busy stream.

iter_chunks

iter_chunks(timeout: Duration | None = None) -> AsyncIterator[Chunk]

Async-iterate raw chunks until the stream ends.

iter_fragments

iter_fragments(timeout: Duration | None = None) -> AsyncIterator[NodeFragment]

Async-iterate raw fragments until the stream ends.

iter_with_deadline

iter_with_deadline(deadline: Time)

Async-iterate deserialized values until deadline or end of stream.

next async

next(obj_type: type[T] | None = None, timeout: Duration | None = None, mimetype_patterns: str | Sequence[str] = '') -> T | Any | None

Alias for next_object: the next deserialized value or None.

Examples:

Process a live audit stream one event at a time:

events.set_expected_types("application/json", AuditEvent)
while (event := await events.next()) is not None:
    await audit_index.store(event)

next_chunk async

next_chunk(timeout: Duration | None = None) -> Chunk | None

Read the next raw chunk, or None at end of stream.

next_fragment async

next_fragment(timeout: Duration | None = None) -> NodeFragment | None

Read the next raw fragment, or None at end of stream.

next_object async

next_object(obj_type: type[T] | None = None, timeout: Duration | None = None, mimetype_patterns: str | Sequence[str] = '') -> T | Any | None

Read and deserialize the next value, or None at end of stream.

put async

put(value: Any, seq: int | None = None, final: bool = False, mimetype: str = '') -> Future[int]

Write value and return its store-confirmation future.

value may be a NodeFragment, a Chunk, or any Python object the node's serialization registry can encode (mimetype selects the encoding). Set final=True on the last data fragment so readers know where the logical value ends. Finality does not close the writer: call drain_and_close after the confirmation future resolves. The returned asyncio.Future resolves to the stored sequence number after the backing store accepts the fragment. Attached WireStream sends are attempted or queued by the writer but are not separately acknowledged.

Examples:

Add an intermediate token while a model response is produced:

await answer.put("The shipment ")

put_chunk async

put_chunk(chunk: Chunk, seq: int | None = None, final: bool = False) -> Future[int]

Admit a native chunk and return its store-confirmation future.

Await this coroutine to respect the writer's bounded admission buffer, then await the returned future when the backing store must have accepted the fragment. Attached stream sends are attempted or queued as the writer processes the batch, but do not add a second delivery confirmation.

put_final async

put_final(value: Any, seq: int | None = None, mimetype: str = '') -> Future[int]

Write value as the logical final element.

This marks the final sequence but leaves the writer open. The returned confirmation can be awaited when immediate store acceptance matters; otherwise a later drain_and_close flushes queued work.

Examples:

Mark the last visible fragment and close the producer:

await answer.put_final("arrives Friday.")

put_fragment async

put_fragment(fragment: NodeFragment) -> Future[int]

Enqueue a NodeFragment (carrying its seq/final).

put_null_final async

put_null_final(seq: int | None = None) -> Future[int]

Write an explicit null fragment as the logical terminator.

Use this after a non-final value when consume should treat that value as one complete unary result. It does not close the writer; finish with drain_and_close after the confirmation resolves.

reset_reader

reset_reader(options: ChunkStoreReaderOptions | dict[str, Any] | None = None) -> AsyncNode

Rewind/reconfigure the reader (e.g. to re-read from an offset).

set_expected_types

set_expected_types(mimetype_patterns: str | Sequence[str], obj_type: type | None) -> AsyncNode

Set the default MIME patterns and object type for reads.

Once set, next()/consume() and async for deserialize to obj_type (matching mimetype_patterns) without repeating those arguments on every call. Returns self for chaining.

set_reader_options

set_reader_options(options: ChunkStoreReaderOptions | dict[str, Any]) -> AsyncNode

Replace the reader options and return self for chaining.

set_serialization_registry

set_serialization_registry(registry: SerializationRegistry) -> AsyncNode

Set the serialization registry and return self for chaining.

set_writer_options

set_writer_options(options: ChunkStoreWriterOptions | dict[str, Any]) -> AsyncNode

Replace the writer options and return self for chaining.

wait_for_buffer_to_drain

wait_for_buffer_to_drain() -> Any

Returns a future that resolves once the write buffer has drained. Await it to apply backpressure from a fast producer, letting consumers catch up before you push more chunks.

NodeMap

NodeMap lets several actions resolve the same named streams. Create one for a related group of actions, then attach it with Action.bind_node_map.

a11.nodes.async_node.NodeMap

NodeMap(chunk_store_factory: Any | None = None)

Creates a node map, optionally backed by a chunk-store factory callable invoked to construct the backing store for each new node.

contains

contains(node_id: str) -> bool

Returns whether a node with the given id exists.

discard

discard(node_id: str, expected: AsyncNode | None = None) -> AsyncNode | None

Removes the node for the given id, optionally only if it matches the expected node.

get

get(node_id: str) -> AsyncNode

Returns the node for the given id, creating it if it does not already exist.

get_if_exists

get_if_exists(node_id: str) -> AsyncNode | None

Returns the node for the given id, or None if it does not exist.

size

size() -> int

Number of nodes in the map.