Actions

An Action is a named, schema-described unit of work whose typed ports are nodes. Actions compose and stream.

Action

Bind local work with bind_handler and start it with run. Input and output ports are live AsyncNode objects throughout the run; wait is the terminal status boundary. Inside a handler, make_nested on Action preserves parent context and call dispatches the child. Bind a Session first with bind_session to make that dispatch remote, and use cancel when the caller no longer needs the result.

a11.actions.action.Action

Action(schema: ActionSchema, action_id: str = '', handler: Any | None = None, *, node_map: NodeMap | None = None, stream: WireStream | None = None, session: Session | None = None, registry: ActionRegistry | None = None, max_concurrent_nested_actions: SupportsInt | SupportsIndex = 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

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()

bind_handler

bind_handler(handler: Any) -> 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: Any | None = None) -> Any

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"].put_final(request.email)
await lookup["email"].drain_and_close()
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() -> Any

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

get_handler

get_handler() -> Any

Return the action's Python handler, or None.

get_header

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

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

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_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() -> Any

Return the action's completion status.

get_stream

get_stream() -> WireStream

Return the action's bound wire stream.

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.

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: Any | None = None) -> Any

Return a future that resolves when the action completes.

wait_for_dispatch

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

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

ActionRegistry

register publishes an async handler under a schema name; make_action then creates a correctly configured action without repeating the schema.

a11.actions.registry.ActionRegistry

ActionRegistry()

Registry mapping action names to their schemas and handlers.

Create an empty action registry.

copy

copy(clear_autofills: bool = True) -> ActionRegistry

Return a copy of the registry, optionally clearing autofills.

get_handler

get_handler(action_name: str) -> Any

Return the Python handler registered under the given action name.

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: Any | 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)

register_sync

register_sync(action_name: str, schema: ActionSchema, handler: Any) -> 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.

Schemas

a11.actions.action.ActionSchema

ActionSchema(name: str, description: str = '', inputs: Any = {}, outputs: Any = {}, headers: Any = {}, 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: Any | None = None)

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

Create a validated port schema.

autofills property writable

autofills: Any

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

description property writable

description: str

Human-readable description of the port.

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: Any

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: Any

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.

Header helpers

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.