Actions¶

An Action is a named, schema-described unit of work whose typed input and output ports are nodes. Actions stream inputs and outputs and execute locally or over remote sessions.

Execution modes and completion¶

An action instance is one-shot. Configure its schema, collaborators, headers, and port mappings before selecting one execution mode:

Mode	Entry point	Execution location
Local	`run()`	A bound handler in the current process
Remote	`call()`	A registered handler reached through a session or wire stream

Both modes expose the same input and output nodes. A caller can start writing a large input after dispatch and read output before the handler finishes. Moving an operation to another process therefore changes its bindings, while its schema and port I/O remain stable.

Starting and finishing are separate barriers. run() schedules a local handler, and call() queues a remote control message. wait_for_dispatch() reports whether a remote peer accepted the call. wait() reports the final handler or remote status after output cleanup. See the Action lifecycle for cancellation, nested calls, and the complete transition model.

Port boundaries¶

Use one port for each result with an independent lifecycle or reader. A model action can stream visible text while returning structured interaction state on another port. An image action can stream progress separately from its finished binary asset. This lets callers drain the outputs concurrently and apply distinct types and size limits.

unary=True declares one complete logical value; other ports carry a sequence of values. The underlying node type remains the same. Handlers mark semantic completion with finalize(). Runtime cleanup closes open writers, but it does not infer that a partial value is complete.

Read or explicitly discard every output a handler can produce. An undrained output can apply backpressure to the producer and delay action completion. The streaming guide covers node reads and the generative media guide shows concurrent outputs with different representations.

Action definitions¶

Schema definitions¶

Actions declare typed ports via ActionSchema and bind execution handlers:

import a11

SCHEMA = a11.ActionSchema(
    name="transform_text",
    description="Transform input text to uppercase.",
    inputs={
        "text": a11.ActionPortSchema(
            name="text", type="text/plain", typeinfo=str, required=True
        )
    },
    outputs={
        "result": a11.ActionPortSchema(
            name="result", type="text/plain", typeinfo=str, required=True
        )
    },
)

async def handler(action: a11.Action) -> None:
    text = await action["text"].consume()
    await action["result"].finalize(text.upper())

action = a11.Action(SCHEMA).bind_handler(handler).run()
await action["text"].finalize("hello world")
print(await action["result"].consume())
await action.wait()

Annotation definitions¶

Declare action handlers directly with typed signatures:

from a11.actions import ActionRegistry

registry = ActionRegistry()

@registry.action(name="summarize")
async def summarize(prompt: str) -> str:
    return f"Summary: {prompt[:50]}..."

A composition of actions needs no signature to read: a flow declares its own ports and is its own handler, so registering one takes the text and nothing else.

greet = registry.flow("""
flow greet {
  in  name:  string
  out reply: string
  "Hello, " then name then "!" -> reply
  drain reply
}
""")

a11.actions.action.Action ¶

Action(schema: ActionSchema, action_id: str = '', handler: ActionHandler | NativeActionHandler | None = None, *, node_map: NodeMap | None = None, stream: WireStream | None = None, session: Session | None = None, registry: ActionRegistry | None = None, max_concurrent_nested_actions: SupportsInt = 64)

A runnable unit of work with typed input/output ports and headers.

Create an action from a schema and optional bindings.

done `property` ¶

done: _DoneEvent

An asyncio.Event-shaped view of completion (await action.done.wait()).

headers `property` ¶

headers: dict[str, bytes]

The action's headers as a mapping of name to bytes.

id `property` `writable` ¶

id: str

The action's id.

schema `property` `writable` ¶

schema: ActionSchema

The action's schema.

settings `property` `writable` ¶

settings: ActionSettings

The action's live ActionSettings (field writes propagate back).

span_id `property` ¶

span_id: str

The action's span id as lowercase hex, or empty when untraced.

trace_id `property` ¶

trace_id: str

The action's trace id as lowercase hex, or empty when untraced.

make_node_id `staticmethod` ¶

make_node_id(action_id: str, node_name: str) -> str

Build the node id for a named port of the given action.

run_in_background `staticmethod` ¶

run_in_background(*args, **kwargs)

run(self: Action) -> Action

Run the action's handler and return the running action. The Python layer also exposes the native entry point as run_in_background.

Examples:

Start local work after binding its handler:

job = Action(SUMMARISE).bind_handler(summarise).run()

add_done_callback ¶

add_done_callback(callback: Callable[['Action'], Any | Awaitable[Any]]) -> Task

Invoke callback(action) once this action completes.

The callback fires exactly once when the action finishes for any reason -- normal completion, a handler error, or cancellation -- because it is driven off the same completion view as done (whose wait swallows the operation status). If the action is already done, the callback still runs on the next event-loop iteration.

A synchronous callback runs to completion; one returning an awaitable is awaited. Cleanup routines registered here should therefore never assume success -- they run on the failure and cancellation paths too, which is exactly what makes this the right hook for releasing resources (e.g. tearing down a transient shell) tied to the action's lifetime.

Returns the scheduled asyncio.Task so the caller can cancel the pending callback or await it. Must be called from within a running event loop.

bind_handler ¶

bind_handler(handler: ActionHandler | NativeActionHandler | None) -> Action

Bind the action's handler and return the action.

bind_node_map ¶

bind_node_map(node_map: NodeMap) -> Action

Bind the action's node map and return the action.

bind_registry ¶

bind_registry(registry: ActionRegistry | None) -> Action

Bind the action's registry and return the action.

bind_session ¶

bind_session(session: Session) -> Action

Bind the action's session and return the action.

bind_stream ¶

bind_stream(stream: WireStream) -> Action

Bind the action's wire stream and return the action.

bind_streams_on_inputs_by_default ¶

bind_streams_on_inputs_by_default(bind: bool) -> Action

Set default stream binding for inputs and return the action.

bind_streams_on_outputs_by_default ¶

bind_streams_on_outputs_by_default(bind: bool) -> Action

Set default stream binding for outputs and return the action.

call ¶

call(wire_headers: Mapping[str, bytes] | None = None) -> Future[Action]

Dispatch the action remotely and return a future of the action.

Examples:

Call a child action and consume its result:

lookup = action.make_nested("find_customer")
await lookup["email"].finalize(request.email)
await lookup.call()
customer = await lookup["customer"].consume(obj_type=Customer)

cancel ¶

cancel() -> None

Cancel the action.

cancelled ¶

cancelled() -> bool

Return True when the action has been cancelled.

clear_inputs_after_run ¶

clear_inputs_after_run(clear: bool = True) -> Action

Set whether inputs are cleared after run and return the action.

clear_outputs_after_run ¶

clear_outputs_after_run(clear: bool = True) -> Action

Set whether outputs are cleared after run and return the action.

contains_port ¶

contains_port(name: str) -> bool

Return True when the action has a port with the given name.

forward_header ¶

forward_header(target: Action, name: str) -> None

Copy a single header from this action to a target action.

forward_headers_with_prefix ¶

forward_headers_with_prefix(target: Action, prefix: str = 'x-a11-') -> None

Copy all headers with the given prefix to a target action.

get_action_message ¶

get_action_message() -> ActionMessage

Return the action's wire message representation.

get_dispatch_status ¶

get_dispatch_status() -> Status | None

Return the action's dispatch status, or None when not dispatched.

get_handler ¶

get_handler() -> ActionHandler | NativeActionHandler | None

Return the action's Python handler, or None.

get_id ¶

get_id() -> str

Return the action's id.

get_input ¶

get_input(name: str, bind_stream: bool | None = None) -> AsyncNode

Return the input port node with the given name.

get_log_node ¶

get_log_node() -> AsyncNode

Return the action's log port node, claiming it for this consumer so the process log sink is not also told.

get_node ¶

get_node(node_id: str) -> AsyncNode

Return the async node with the given id.

get_node_map ¶

get_node_map() -> NodeMap

Return the action's bound node map.

get_output ¶

get_output(name: str, bind_stream: bool | None = None) -> AsyncNode

Return the output port node with the given name.

get_port ¶

get_port(name: str) -> AsyncNode

Return the port node with the given name.

get_registry ¶

get_registry() -> ActionRegistry

Return the action's bound registry.

get_schema ¶

get_schema() -> ActionSchema

Return the action's schema.

get_session ¶

get_session() -> Session

Return the action's bound session.

get_status ¶

get_status() -> Status

Return the action's completion status.

get_stream ¶

get_stream() -> WireStream

Return the action's bound wire stream.

get_verified_authorization ¶

get_verified_authorization() -> VerifiedAuthorization

Return identity and authority already verified for this action.

has_been_called ¶

has_been_called() -> bool

Return True when the action has been dispatched remotely.

has_been_run ¶

has_been_run() -> bool

Return True when the action has been run locally.

has_handler ¶

has_handler() -> bool

Return True when the action has a handler bound.

has_header ¶

has_header(name: str) -> bool

Return True when the action has a header with the given name.

is_done ¶

is_done() -> bool

Return True when the action has finished.

log `async` ¶

log(value: Any, *, level: str | None = None, mimetype: str | None = None, metadata: Mapping[str, bytes | str] | None = None, channel: str | None = None, internal: bool = False, file: str | None = None, lineno: int | None = None) -> None

Log value on the action's reserved log port.

The object becomes a chunk exactly as node.put(value) would make one -- a str is text/plain, bytes are application/octet-stream, anything else is JSON -- and the chunk always carries a timestamp. Pass an already-built Chunk to log it as it is.

The reserved log port requires no schema declaration or manual drain. The action closes it with its other outputs and creates it only when used. Only a running action may log; logging before run or from the calling side of call raises.

A nested action forwards unclaimed logs through its parent. A root action sends them to A11's logger (see a11.logging). Call get_log_node before the action runs to iterate over its chunks and suppress the default route.

Parameters:

Name	Type	Description	Default
`value`	`Any`	What to log.	required
`level`	`str \| None`	One of `LOG_LEVELS`; `None` is `"info"`.	`None`
`mimetype`	`str \| None`	Media-type hint for the serializer.	`None`
`metadata`	`Mapping[str, bytes \| str] \| None`	Extra chunk attributes, merged before the arguments above -- so an explicit `level` wins over one written here.	`None`
`channel`	`str \| None`	A label a consumer can filter on.	`None`
`internal`	`bool`	Whether this is A11's own bookkeeping rather than something an end user should be shown.	`False`
`file`	`str \| None`	Source file to report; defaults to the caller's.	`None`
`lineno`	`int \| None`	Source line to report; defaults to the caller's.	`None`

log_chunk ¶

log_chunk(chunk: Chunk, level: str | None = None, metadata: Mapping[str, str] | None = None, channel: str | None = None, file: str | None = None, lineno: SupportsInt | None = None, internal: bool = False) -> None

Write an already-built chunk to the action's log port. Only a running action may log.

logf `async` ¶

logf(format: str, *args: Any, level: str | None = None, metadata: Mapping[str, bytes | str] | None = None, channel: str | None = None, internal: bool = False, file: str | None = None, lineno: int | None = None) -> None

Log format % args -- percent-style, as logging formats.

await action.logf("read %d of %d pages", done, total). The interpolation happens here rather than in a handler, so a message whose level is filtered out still costs only the call. Everything else is log's.

map_ports_from_message ¶

map_ports_from_message(message: ActionMessage) -> Action

Map the action's ports from a wire message and return the action.

remove_header ¶

remove_header(name: str) -> None

Remove the header with the given name.

run ¶

run() -> Action

Run the action's handler and return the running action. The Python layer also exposes the native entry point as run_in_background.

Examples:

Start local work after binding its handler:

job = Action(SUMMARISE).bind_handler(summarise).run()

set_header ¶

set_header(name: str, value: Any) -> Action

Set a header from a str or bytes value and return the action.

set_id ¶

set_id(action_id: str) -> Action

Set the action's id and return the action.

set_on_cancelled ¶

set_on_cancelled(callback: Any) -> None

Register a synchronous callback invoked when the action is cancelled.

set_schema ¶

set_schema(schema: ActionSchema) -> Action

Set the action's schema and return the action.

set_span_attribute ¶

set_span_attribute(key: str, value: Any) -> None

Set an attribute on the action's span; no-op when untraced.

set_span_input ¶

set_span_input(value: Any) -> None

Record this action span's input (Langfuse observation input).

set_span_name ¶

set_span_name(name: str) -> None

Set the name of the action's span.

set_span_output ¶

set_span_output(value: Any) -> None

Record this action span's output (Langfuse observation output).

set_span_status ¶

set_span_status(code: str, description: str = '') -> None

Set the span status explicitly ('ok', 'error' or 'unset').

wait ¶

wait(timeout: Duration | None = None) -> Future[Action]

Return a future that resolves when the action completes.

wait_for_dispatch ¶

wait_for_dispatch(timeout: Duration | None = None) -> Future[Status]

Return a future that resolves when the action has been dispatched.

get_header ¶

get_header(name: str, decode: Literal[False] = False) -> bytes | None

get_header(name: str, decode: Literal[True] = True) -> str | None

get_header(name: str, decode: bool = False) -> bytes | str | None

get_header(name: str, decode: Literal[False] = False) -> bytes | None

Return header name (None if absent); decode UTF-8 to str.

make_nested ¶

make_nested(schema: ActionSchema, propagate_io: bool = True, forward_headers: bool = True) -> Action

make_nested(action_name: str, propagate_io: bool = True, forward_headers: bool = True) -> Action

make_nested(schema: ActionSchema, propagate_io: bool = True, forward_headers: bool = True) -> Action

Create a nested child action from a schema.

ActionRegistry¶

ActionRegistry manages action schemas and handlers for local execution and dispatch through networked services. The schema provides discovery and validation; the handler supplies the local implementation. A registry can expose the same contract to application calls, LLM tools, remote peers, and Flow compositions.

registry = ActionRegistry()
registry.register("transform_text", SCHEMA, handler)

action = registry.make_action("transform_text")

a11.actions.registry.ActionRegistry ¶

ActionRegistry()

Registry mapping action names to their schemas and handlers.

Create an empty action registry.

action ¶

action(fn: Callable[..., Any] | None = None, *, name: str | None = None, description: str | None = None, output: Any = 'output', headers: Mapping[str, ActionHeaderSchema] | None = None) -> Callable[..., Any]

Register a function as an Action, deriving both halves from it.

The decorator form of action_from_callable: the schema and the handler are built from the function's annotations and registered here, so the whole Action is the function and its signature.

registry = ActionRegistry()

@registry.action
async def summarise(
    document: str,
    style: Annotated[str | None, a11.InputPort(description="Tone.")] = None,
) -> str:
    # The docstring becomes the Action's description.
    return await model.summarise(document, style or "neutral")

The function comes back unchanged, so it stays directly callable and testable; what was derived from it is on fn.action_schema and fn.action_handler.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Any] \| None`	The function to bind, when used bare as `@registry.action`.	`None`
`name`	`str \| None`	Action name to register under; defaults to `fn.__name__`.	`None`
`description`	`str \| None`	Action description; defaults to `fn`'s docstring.	`None`
`output`	`Any`	Where `fn`'s own result goes -- a port name, or an `OutputPort` to say more about that port. Ignored by a function that declares `OutputPort` parameters, which has no such result.	`'output'`
`headers`	`Mapping[str, ActionHeaderSchema] \| None`	Extra header schemas to merge into the Action's, for headers the function does not itself take a parameter for.	`None`

Returns:

Type	Description
`Callable[..., Any]`	`fn` itself, or the decorator that will take it.

copy ¶

copy(clear_autofills: bool = True) -> ActionRegistry

Return a copy of the registry, optionally clearing autofills.

flow ¶

flow(source: str, *, name: str | None = None, source_name: str = '') -> FlowPlan

Register a Flow composition as an Action, deriving both halves from it.

What action is for a function, this is for a flow -- except that there is nothing to infer from Python here. A flow declares its own ports and is its own handler, so the text is the whole Action and this takes it as it stands:

registry = ActionRegistry()

greet = registry.flow('''
    flow greet {
      in  name:  string
      out reply: string
      "Hello, " then name then "!" -> reply
      drain reply
    }
''')

The source must declare exactly one flow, and a named one: a file of several is register_all's business, and a flow { ... } entry point has no name to be called by.

Parameters:

Name	Type	Description	Default
`source`	`str`	The flow's text.	required
`name`	`str \| None`	Action name to register under; defaults to the flow's own. The schema is registered under this name too, so a peer asking what this side serves is told the name it can actually call.	`None`
`source_name`	`str`	What to call the source in a diagnostic -- a file name, usually. Defaults to the name it is registered under.	`''`

Returns:

Type	Description
`FlowPlan`	The compiled FlowPlan, which is also how to
`FlowPlan`	run the composition here rather than through the registry.

Raises:

Type	Description
`FlowSyntaxError`	If the source will not compile.
`ValueError`	If it is not exactly one named flow.

get_handler ¶

get_handler(action_name: str) -> ActionHandler | NativeActionHandler | None

Return the handler registered under the given action name: the Python callable it was registered with, a NativeActionHandler when the action is implemented in C++, or None when it has no handler.

get_schema ¶

get_schema(action_name: str) -> ActionSchema

Return the schema registered under the given action name.

is_registered ¶

is_registered(action_name: str) -> bool

Return True when an action with the given name is registered.

list_registered_actions ¶

list_registered_actions() -> list[str]

Return the names of all registered actions.

make_action ¶

make_action(action_name: str, action_id: str = '', node_map: NodeMap | None = None, stream: WireStream | None = None, session: Session | None = None) -> Action

Create an action instance from a registered action name.

Examples:

Construct and start work without repeating the schema:

job = registry.make_action("summarise")
job.run()

make_action_message ¶

make_action_message(action_name: str, action_id: str = '') -> ActionMessage

Create a wire action message for a registered action name.

register ¶

register(action_name: str, schema: ActionSchema, handler: ActionHandler | NativeActionHandler | None = None) -> None

Register an action with a schema and optional async handler.

Examples:

Publish an application handler under its schema name:

registry.register("summarise", SUMMARISE, summarise)

Or the schema alone, which says the action lives on a peer and is to be reached with a flow's call rather than run here:

registry.register("shell_execute", SHELL_EXECUTE)

Each port carrying a typeinfo also gets a json_schema derived from it on the way in, because only Python can read a Python type. That happens here rather than when the action is described, so the answer a peer gets over the wire -- built by the native describer, which sees only what is on the schema -- is the same document a local caller gets. See a11.actions.describe.

register_sync ¶

register_sync(action_name: str, schema: ActionSchema, handler: ActionHandler | NativeActionHandler | None) -> None

Register an action with a schema and a synchronous handler.

unregister ¶

unregister(action_name: str) -> None

Remove the action with the given name from the registry.

Actions from annotations¶

a11.actions.annotated ¶

Build an Action's schema and handler from a callable's annotations.

The same trade FastAPI makes for request handlers: write an ordinary function whose parameters are the values it actually wants, and let the framework work out the wire shape from the annotations and do the marshalling.

async def summarise(
    document: str,
    style: Annotated[str | None, InputPort(description="Tone to use.")] = None,
) -> str:
    # Whatever the docstring says becomes the Action's description.
    return model.summarise(document, style or "neutral")

schema, handler = action_from_callable(summarise)

That declares a summarise Action with a required unary document input, an optional unary style input, and one unary output port; the handler consumes the inputs, calls the function, and finalises its return value onto the output. ActionRegistry.action is the same thing as a decorator that registers the result.

How a parameter is read¶

Each parameter is classified once, when the Action is built:

Action -- the running action itself, the way FastAPI hands over its Request. Nothing is declared for it.
Annotated with Header -- an action header, decoded to the parameter's type.
Annotated with OutputPort -- an output port, handed over as a live AsyncNode for the function to write itself.
anything else -- an input port, optionally annotated with InputPort to say more about it.

An input's arity is its annotation: AsyncIterator[T] (or AsyncIterable/AsyncGenerator) is a stream and the parameter receives a lazy async iterator of T; anything else is unary and the parameter receives one value. A unary input is required unless its type admits None (T | None, Optional[T]) or the parameter has a default. Annotate a parameter AsyncNode to be handed the node itself instead of decoded values.

Every declared name defaults to the parameter's name, with a header turning underscores into hyphens -- so x_a11_shell_id is the x-a11-shell-id header -- and every one can be overridden on the annotation.

How the outputs are decided¶

If no parameter is annotated with OutputPort, the function's own result is the Action's single output, named output unless action_from_callable's output argument says otherwise:

an async def returning T gets a unary port, finalised with the returned value (a None from a T | None return finalises the port empty);
an async generator gets a streaming port, each yielded value written to it in turn;
an async def annotated -> None declares no output at all.

A function that wants several outputs takes them as parameters instead, and must then return nothing:

async def split(
    text: str,
    words: Annotated[AsyncNode, OutputPort(description="One word each.")],
    total: Annotated[AsyncNode, OutputPort(mimetype="application/json")],
) -> None:
    for word in text.split():
        await words.put(word)
    await words.finalize()
    await total.finalize(len(text.split()))

Those nodes are ordinary ones: write, finalize, close, or leave them to the runner, which closes any output the handler did not write.

What a caller owes¶

A11's client contract already asks a caller to close every input port it declares, and a derived handler needs that too: an input port left open with nothing in it is indistinguishable from one whose value has not arrived yet, so reading it waits. The action's x-a11-deadline bounds that wait. A caller that neither fills nor closes an optional input and sets no deadline waits indefinitely.

InputPort `dataclass` ¶

InputPort(*, name: str | None = None, mimetype: str | None = None, description: str = '', typeinfo: type | None = None, required: bool | None = None, unary: bool | None = None, autofills: Sequence[Any] | None = None)

Describe the input port a parameter is read from.

Every field is optional: what is left out is inferred from the parameter (its name, its type, whether it admits None). Attach one with Annotated[T, InputPort(...)].

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Port name; defaults to the parameter's name.	`None`
`mimetype`	`str \| None`	Declared media type; defaults to the one the parameter's type serialises as (`text/plain` for `str`, `application/octet-stream` for `bytes`, `application/json` otherwise).	`None`
`description`	`str`	What the port carries. Reaches an LLM that is offered this Action as a tool, so write it for that reader.	`''`
`typeinfo`	`type \| None`	Python type published on the schema; defaults to the parameter's own type when it is a plain class.	`None`
`required`	`bool \| None`	Whether a caller must supply the port. Defaults to `False` for a parameter that admits `None` or has a default, and for streams, and `True` otherwise.	`None`
`unary`	`bool \| None`	Whether the port carries a single whole value. Defaults to the parameter's arity and normally needs no override.	`None`
`autofills`	`Sequence[Any] \| None`	Fragments the runtime fills the port with, as on `ActionPortSchema`.	`None`

OutputPort `dataclass` ¶

OutputPort(*, name: str | None = None, mimetype: str | None = None, description: str = '', typeinfo: type | None = None, required: bool | None = None, unary: bool | None = None, autofills: Sequence[Any] | None = None)

Declare an output port handed to the function as an AsyncNode.

Annotate a parameter Annotated[AsyncNode, OutputPort(...)] and the handler passes the live output node, which makes the function responsible for what goes on it. A function that declares any of these must return None: its own result no longer has a port to go to.

The fields are InputPort's, read the same way, except that typeinfo and mimetype have no parameter type to be inferred from -- a node is just a node -- so an undeclared mimetype is application/json.

Header `dataclass` ¶

Header(*, name: str | None = None, description: str = '', default: Any = UNSET, default_factory: Callable[[], Any] | None = None)

Describe the action header a parameter is read from.

Attach one with Annotated[T, Header(...)]. The raw header bytes are decoded to the parameter's type: bytes verbatim, str as UTF-8, anything else validated by Pydantic from the value as JSON and then, if that is not JSON, from the decoded text -- so int, an enum, a model, and a list[str] all work.

A header the caller omitted uses default, or default_factory(), or is None if the parameter admits it; with none of those the action fails with INVALID_ARGUMENT, which is the only sense in which a header is "required". A static default is also published on the schema, so the runtime seeds it into the action's headers and a nested call inherits it.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Header name; defaults to the parameter's name with underscores turned into hyphens.	`None`
`description`	`str`	What the header means, for whoever calls the Action.	`''`
`default`	`Any`	Value used when the header is absent.	`UNSET`
`default_factory`	`Callable[[], Any] \| None`	Called for that value instead, for one that must not be shared (or cannot be computed until the call).	`None`

action_from_callable ¶

action_from_callable(fn: Callable[..., Any], *, name: str | None = None, description: str | None = None, output: str | OutputPort = DEFAULT_OUTPUT_NAME, headers: Mapping[str, ActionHeaderSchema] | None = None) -> tuple[ActionSchema, ActionHandler]

Derive an Action's schema and handler from fn's annotations.

See the module documentation for how each parameter and the return type are read. Everything is worked out here, once: the returned handler only marshals.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Any]`	The function to bind. Usually `async def`, either returning a value or an async generator yielding them; a plain `def` is accepted too.	required
`name`	`str \| None`	Action name; defaults to `fn.__name__`.	`None`
`description`	`str \| None`	Action description; defaults to `fn`'s docstring.	`None`
`output`	`str \| OutputPort`	Where `fn`'s own result goes -- a port name, or an `OutputPort` to say more about it. Ignored by a function that declares `OutputPort` parameters, which has no such result.	`DEFAULT_OUTPUT_NAME`
`headers`	`Mapping[str, ActionHeaderSchema] \| None`	Extra header schemas to merge in, for headers the function does not itself take a parameter for. Pass `DEFAULT_ACTION_HEADERS` to declare A11's own.	`None`

Returns:

Type	Description
`ActionSchema`	The schema and the handler, in the order
`ActionHandler`	`register` takes them.

Raises:

Type	Description
`TypeError`	If an annotation cannot be turned into a port -- an unannotated parameter, two parameters claiming one name, a function that both declares outputs and returns a value.

a11.actions.annotated.action_from_callable ¶

action_from_callable(fn: Callable[..., Any], *, name: str | None = None, description: str | None = None, output: str | OutputPort = DEFAULT_OUTPUT_NAME, headers: Mapping[str, ActionHeaderSchema] | None = None) -> tuple[ActionSchema, ActionHandler]

Derive an Action's schema and handler from fn's annotations.

See the module documentation for how each parameter and the return type are read. Everything is worked out here, once: the returned handler only marshals.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Any]`	The function to bind. Usually `async def`, either returning a value or an async generator yielding them; a plain `def` is accepted too.	required
`name`	`str \| None`	Action name; defaults to `fn.__name__`.	`None`
`description`	`str \| None`	Action description; defaults to `fn`'s docstring.	`None`
`output`	`str \| OutputPort`	Where `fn`'s own result goes -- a port name, or an `OutputPort` to say more about it. Ignored by a function that declares `OutputPort` parameters, which has no such result.	`DEFAULT_OUTPUT_NAME`
`headers`	`Mapping[str, ActionHeaderSchema] \| None`	Extra header schemas to merge in, for headers the function does not itself take a parameter for. Pass `DEFAULT_ACTION_HEADERS` to declare A11's own.	`None`

Returns:

Type	Description
`ActionSchema`	The schema and the handler, in the order
`ActionHandler`	`register` takes them.

Raises:

Type	Description
`TypeError`	If an annotation cannot be turned into a port -- an unannotated parameter, two parameters claiming one name, a function that both declares outputs and returns a value.

a11.actions.annotated.InputPort `dataclass` ¶

InputPort(*, name: str | None = None, mimetype: str | None = None, description: str = '', typeinfo: type | None = None, required: bool | None = None, unary: bool | None = None, autofills: Sequence[Any] | None = None)

Describe the input port a parameter is read from.

Every field is optional: what is left out is inferred from the parameter (its name, its type, whether it admits None). Attach one with Annotated[T, InputPort(...)].

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Port name; defaults to the parameter's name.	`None`
`mimetype`	`str \| None`	Declared media type; defaults to the one the parameter's type serialises as (`text/plain` for `str`, `application/octet-stream` for `bytes`, `application/json` otherwise).	`None`
`description`	`str`	What the port carries. Reaches an LLM that is offered this Action as a tool, so write it for that reader.	`''`
`typeinfo`	`type \| None`	Python type published on the schema; defaults to the parameter's own type when it is a plain class.	`None`
`required`	`bool \| None`	Whether a caller must supply the port. Defaults to `False` for a parameter that admits `None` or has a default, and for streams, and `True` otherwise.	`None`
`unary`	`bool \| None`	Whether the port carries a single whole value. Defaults to the parameter's arity and normally needs no override.	`None`
`autofills`	`Sequence[Any] \| None`	Fragments the runtime fills the port with, as on `ActionPortSchema`.	`None`

a11.actions.annotated.OutputPort `dataclass` ¶

OutputPort(*, name: str | None = None, mimetype: str | None = None, description: str = '', typeinfo: type | None = None, required: bool | None = None, unary: bool | None = None, autofills: Sequence[Any] | None = None)

Declare an output port handed to the function as an AsyncNode.

Annotate a parameter Annotated[AsyncNode, OutputPort(...)] and the handler passes the live output node, which makes the function responsible for what goes on it. A function that declares any of these must return None: its own result no longer has a port to go to.

The fields are InputPort's, read the same way, except that typeinfo and mimetype have no parameter type to be inferred from -- a node is just a node -- so an undeclared mimetype is application/json.

a11.actions.annotated.Header `dataclass` ¶

Header(*, name: str | None = None, description: str = '', default: Any = UNSET, default_factory: Callable[[], Any] | None = None)

Describe the action header a parameter is read from.

Attach one with Annotated[T, Header(...)]. The raw header bytes are decoded to the parameter's type: bytes verbatim, str as UTF-8, anything else validated by Pydantic from the value as JSON and then, if that is not JSON, from the decoded text -- so int, an enum, a model, and a list[str] all work.

A header the caller omitted uses default, or default_factory(), or is None if the parameter admits it; with none of those the action fails with INVALID_ARGUMENT, which is the only sense in which a header is "required". A static default is also published on the schema, so the runtime seeds it into the action's headers and a nested call inherits it.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Header name; defaults to the parameter's name with underscores turned into hyphens.	`None`
`description`	`str`	What the header means, for whoever calls the Action.	`''`
`default`	`Any`	Value used when the header is absent.	`UNSET`
`default_factory`	`Callable[[], Any] \| None`	Called for that value instead, for one that must not be shared (or cannot be computed until the call).	`None`

Schemas¶

a11.actions.action.ActionSchema ¶

ActionSchema(name: str, description: str = '', inputs: Any = {}, outputs: Any = {}, headers: Mapping[str, bytes] | None = {}, output_to_json_field: Mapping[str, str] = {})

Schema describing an action's ports, headers and output mappings.

Create a validated action schema.

description `property` `writable` ¶

description: str

Human-readable description of the action.

headers `property` `writable` ¶

headers: _ActionHeaderSchemaMapView

Mapping of header names to their header schemas.

inputs `property` `writable` ¶

inputs: _ActionPortSchemaMapView

Mapping of input port names to their port schemas.

name `property` `writable` ¶

name: str

The action's name.

output_to_json_field `property` `writable` ¶

output_to_json_field: _StringSchemaMapView

Mapping of output port names to JSON field names.

outputs `property` `writable` ¶

outputs: _ActionPortSchemaMapView

Mapping of output port names to their port schemas.

map_output_to_json ¶

map_output_to_json(output_name: str, field_name: str = '') -> None

Map an output port to a JSON field in the action's response.

validate ¶

validate() -> None

Validate the action schema, raising on error.

a11.actions.action.ActionPortSchema ¶

ActionPortSchema(name: str, type: str, description: str = '', required: bool = False, unary: bool = False, autofills: Any | None = None, typeinfo: type | None = None, json_schema: str = '')

Schema describing a single input or output port of an action.

Create a validated port schema.

autofills `property` `writable` ¶

autofills: list[NodeFragment | None] | None

Node fragments used to autofill the port, or None entries.

description `property` `writable` ¶

description: str

Human-readable description of the port.

json_schema `property` `writable` ¶

json_schema: str

JSON Schema for the port's payload, as text. The describable half of typeinfo: what this side can say about the type to a peer or a model, which a descriptor that crossed a wire has no type handle left to derive.

name `property` `writable` ¶

name: str

The port's name.

required `property` `writable` ¶

required: bool

Whether the port must be provided.

type `property` `writable` ¶

type: str

The port's data type name.

typeinfo `property` `writable` ¶

typeinfo: type | None

Optional Python type associated with the port, or None.

unary `property` `writable` ¶

unary: bool

Whether the port carries a single value.

validate ¶

validate() -> None

Validate the port schema, raising on error.

a11.actions.action.ActionHeaderSchema ¶

ActionHeaderSchema(name: str, description: str = '', default: Any | None = None)

Schema describing a single header of an action.

Create a validated header schema.

default `property` `writable` ¶

default: bytes | None

Default header value as bytes, or None when unset.

description `property` `writable` ¶

description: str

Human-readable description of the header.

name `property` `writable` ¶

name: str

The header's name.

validate ¶

validate() -> None

Validate the header schema, raising on error.

a11.actions.action.ActionSettings ¶

ActionSettings(bind_streams_on_inputs_by_default: bool | None = None, bind_streams_on_outputs_by_default: bool | None = None, clear_inputs_after_run: bool = False, clear_outputs_after_run: bool = False)

Runtime settings controlling an action's stream binding and cleanup.

Create action settings.

Describing actions¶

a11.actions.describe ¶

An ActionSchema in JSON, which is how one travels.

One concept in two representations, and this is the crossing between them. An ActionSchema is the live object; a port's typeinfo is a Python type and an input's autofills are receiver-owned values, so neither can go on a wire. The a11.actions/v1 document is the same schema written as text, with typeinfo replaced by the port's json_schema and the autofill values by an autofilled flag. Its shape lives in cpp/a11/actions/describe.h; these are thin wrappers over it.

The one field a document carries that a schema does not is runnable, which says whether the answering side holds a handler. It is the registry's annotation on a schema rather than part of one -- the same schema is runnable here and schema-only there, and that difference is what Flow reads to choose run over call.

What Python adds is the one thing C++ cannot do: derive a port's json_schema from its typeinfo, which only a Python runtime can read. That happens at registration, not when a schema is written out, and that is the whole design. A peer asking __list_actions__ is answered by the native writer, which sees only what is on the schema; if the derivation happened in a Python path instead, a schema written locally would carry types and the same schema written for a peer would not. Deriving once, when the action becomes discoverable, is what makes those two answers the same document.

json_schema_for ¶

json_schema_for(port: ActionPortSchema) -> str

A JSON Schema for port's payload, as text, or "".

Empty when the port has no typeinfo to derive from. A port with no stated type is described without one, and the adapter presents it to a model as {"type": "object"}.

fill_json_schemas ¶

fill_json_schemas(schema: ActionSchema) -> ActionSchema

Derives json_schema for every port of schema that lacks one.

Mutates and returns schema. Idempotent: a port that already has one is left alone, so a caller that stated a schema by hand keeps theirs.

schema_to_json ¶

schema_to_json(schema: ActionSchema, *, runnable: bool = True, all_ports: bool = False) -> dict[str, Any]

One action's a11.actions/v1 entry.

The entry rather than the envelope: a caller with one schema in hand wants the thing that goes in an actions array, and registry_to_json is what produces a whole document. The native call returns the envelope so that the HTTP endpoint's item route and its collection route have the same shape; this unwraps it.

Parameters:

Name	Type	Description	Default
`schema`	`ActionSchema`	The action's interface.	required
`runnable`	`bool`	Whether this side holds a handler for it. A schema registered without one means "this action lives on the peer".	`True`
`all_ports`	`bool`	Keep inputs the receiver autofills, flagged. A caller cannot write them, so they are omitted by default.	`False`

registry_to_json ¶

registry_to_json(registry: ActionRegistry, request: Any = None) -> dict[str, Any]

Every action in registry as one a11.actions/v1 document.

Parameters:

Name	Type	Description	Default
`registry`	`ActionRegistry`	The registry to describe.	required
`request`	`Any`	What `__list_actions__` takes on its `request` port -- a mapping with any of `names` (full-match patterns), `exact`, `ports` (`"callable"` or `"all"`), `include_reserved` and `runnable_only`; or a bare list of patterns; or None for all of them.	`None`

schemas_in_document ¶

schemas_in_document(document: Any) -> list[dict[str, Any]]

The entries of a document, accepting the envelope or a bare list.

A caller handed one or the other -- a whole document from __list_actions__ or just its entries -- should not have to care which.

schema_from_json ¶

schema_from_json(described: dict[str, Any]) -> ActionSchema

Rebuilds an ActionSchema from one described action.

For a side that has to call what it was told about: a tool bridge registering a reverse-dispatch proxy, or a flow run against a peer registering the peer's actions for their schemas alone.

A port's local typeinfo handle returns as None. Receiver-owned autofill defaults remain local, while json_schema is preserved for remote tool argument types.

builtin_action_names ¶

builtin_action_names() -> list[str]

The actions every registry answers for, whatever it was built to do.

is_reserved_action ¶

is_reserved_action(name: str) -> bool

Whether name is one of A11's own, rather than an application's.

The __-prefix rule, which replaces the hand-maintained exclusion lists that each discovery workaround kept for itself.

a11.actions.describe.schema_to_json ¶

schema_to_json(schema: ActionSchema, *, runnable: bool = True, all_ports: bool = False) -> dict[str, Any]

One action's a11.actions/v1 entry.

The entry rather than the envelope: a caller with one schema in hand wants the thing that goes in an actions array, and registry_to_json is what produces a whole document. The native call returns the envelope so that the HTTP endpoint's item route and its collection route have the same shape; this unwraps it.

Parameters:

Name	Type	Description	Default
`schema`	`ActionSchema`	The action's interface.	required
`runnable`	`bool`	Whether this side holds a handler for it. A schema registered without one means "this action lives on the peer".	`True`
`all_ports`	`bool`	Keep inputs the receiver autofills, flagged. A caller cannot write them, so they are omitted by default.	`False`

a11.actions.describe.schema_from_json ¶

schema_from_json(described: dict[str, Any]) -> ActionSchema

Rebuilds an ActionSchema from one described action.

For a side that has to call what it was told about: a tool bridge registering a reverse-dispatch proxy, or a flow run against a peer registering the peer's actions for their schemas alone.

A port's local typeinfo handle returns as None. Receiver-owned autofill defaults remain local, while json_schema is preserved for remote tool argument types.

a11.actions.describe.registry_to_json ¶

registry_to_json(registry: ActionRegistry, request: Any = None) -> dict[str, Any]

Every action in registry as one a11.actions/v1 document.

Parameters:

Name	Type	Description	Default
`registry`	`ActionRegistry`	The registry to describe.	required
`request`	`Any`	What `__list_actions__` takes on its `request` port -- a mapping with any of `names` (full-match patterns), `exact`, `ports` (`"callable"` or `"all"`), `include_reserved` and `runnable_only`; or a bare list of patterns; or None for all of them.	`None`

a11.actions.describe.schemas_in_document ¶

schemas_in_document(document: Any) -> list[dict[str, Any]]

The entries of a document, accepting the envelope or a bare list.

A caller handed one or the other -- a whole document from __list_actions__ or just its entries -- should not have to care which.

a11.actions.describe.fill_json_schemas ¶

fill_json_schemas(schema: ActionSchema) -> ActionSchema

Derives json_schema for every port of schema that lacks one.

Mutates and returns schema. Idempotent: a port that already has one is left alone, so a caller that stated a schema by hand keeps theirs.

a11.actions.describe.json_schema_for ¶

json_schema_for(port: ActionPortSchema) -> str

A JSON Schema for port's payload, as text, or "".

Empty when the port has no typeinfo to derive from. A port with no stated type is described without one, and the adapter presents it to a model as {"type": "object"}.

a11.actions.describe.builtin_action_names ¶

builtin_action_names() -> list[str]

The actions every registry answers for, whatever it was built to do.

a11.actions.describe.is_reserved_action ¶

is_reserved_action(name: str) -> bool

Whether name is one of A11's own, rather than an application's.

The __-prefix rule, which replaces the hand-maintained exclusion lists that each discovery workaround kept for itself.

Header helpers¶

Signed delegation proofs and connection-scoped authorization contexts use the reserved action headers described in the authorization reference.

a11.actions.action.DefaultHeaders ¶

Bases: StrEnum

Well-known action metadata understood by A11 integrations.

Headers describe one call and normally flow into nested actions. Use these names instead of ad-hoc equivalents so deadlines, tool policy, user logs, and tracing remain connected across agent boundaries.

a11.actions.action.DEFAULT_HEADERS `module-attribute` ¶

DEFAULT_HEADERS = {DefaultHeaders.DEADLINE: ActionHeaderSchema(DefaultHeaders.DEADLINE, 'Deadline for execution in milliseconds since epoch.'), DefaultHeaders.AUTH: ActionHeaderSchema(DefaultHeaders.AUTH, 'Signed A11 subject and caller delegation chain.'), DefaultHeaders.ALLOWED_LLM_ACTIONS: ActionHeaderSchema(DefaultHeaders.ALLOWED_LLM_ACTIONS, 'Comma-separated regex patterns of actions the LLM may call as tools.'), DefaultHeaders.OTEL_TRACEPARENT: ActionHeaderSchema(DefaultHeaders.OTEL_TRACEPARENT, 'OpenTelemetry traceparent header.'), DefaultHeaders.OTEL_TRACESTATE: ActionHeaderSchema(DefaultHeaders.OTEL_TRACESTATE, 'OpenTelemetry tracestate header.'), DefaultHeaders.OTEL_BAGGAGE: ActionHeaderSchema(DefaultHeaders.OTEL_BAGGAGE, 'OpenTelemetry baggage header.')}

Actions¶

Execution modes and completion¶

Port boundaries¶

Action definitions¶

Schema definitions¶

Annotation definitions¶

a11.actions.action.Action ¶

done property ¶

headers property ¶

id property writable ¶

schema property writable ¶

settings property writable ¶

span_id property ¶

trace_id property ¶

make_node_id staticmethod ¶

run_in_background staticmethod ¶

add_done_callback ¶

bind_handler ¶

bind_node_map ¶

bind_registry ¶

bind_session ¶

bind_stream ¶

bind_streams_on_inputs_by_default ¶

bind_streams_on_outputs_by_default ¶

call ¶

cancel ¶

cancelled ¶

clear_inputs_after_run ¶

clear_outputs_after_run ¶

contains_port ¶

forward_header ¶

forward_headers_with_prefix ¶

get_action_message ¶

get_dispatch_status ¶

get_handler ¶

get_id ¶

get_input ¶

get_log_node ¶

get_node ¶

get_node_map ¶

get_output ¶

get_port ¶

get_registry ¶

get_schema ¶

get_session ¶

get_status ¶

get_stream ¶

get_verified_authorization ¶

has_been_called ¶

has_been_run ¶

has_handler ¶

has_header ¶

is_done ¶

log async ¶

log_chunk ¶

logf async ¶

map_ports_from_message ¶

remove_header ¶

run ¶

set_header ¶

set_id ¶

set_on_cancelled ¶

set_schema ¶

set_span_attribute ¶

set_span_input ¶

set_span_name ¶

set_span_output ¶

set_span_status ¶

wait ¶

wait_for_dispatch ¶

get_header ¶

make_nested ¶

ActionRegistry¶

a11.actions.registry.ActionRegistry ¶

action ¶

copy ¶

flow ¶

get_handler ¶

get_schema ¶

is_registered ¶

done `property` ¶

headers `property` ¶

id `property` `writable` ¶

schema `property` `writable` ¶

settings `property` `writable` ¶

span_id `property` ¶

trace_id `property` ¶

make_node_id `staticmethod` ¶

run_in_background `staticmethod` ¶

log `async` ¶

logf `async` ¶

InputPort `dataclass` ¶

OutputPort `dataclass` ¶

Header `dataclass` ¶

a11.actions.annotated.InputPort `dataclass` ¶

a11.actions.annotated.OutputPort `dataclass` ¶

a11.actions.annotated.Header `dataclass` ¶

description `property` `writable` ¶

headers `property` `writable` ¶

inputs `property` `writable` ¶

name `property` `writable` ¶

output_to_json_field `property` `writable` ¶

outputs `property` `writable` ¶

autofills `property` `writable` ¶

description `property` `writable` ¶

json_schema `property` `writable` ¶

name `property` `writable` ¶

required `property` `writable` ¶

type `property` `writable` ¶

typeinfo `property` `writable` ¶

unary `property` `writable` ¶

default `property` `writable` ¶

description `property` `writable` ¶

name `property` `writable` ¶

a11.actions.action.DEFAULT_HEADERS `module-attribute` ¶