axio

Core agent loop, protocols, messages, tools, events, and context interfaces.

Public API

axio - public API.

class axio.Agent(system: 'str', transport: 'CompletionTransport', tools: 'list[Tool[Any]]' = <factory>, selector: 'ToolSelector | None' = None, max_iterations: 'int' = 50, last_iteration_message: 'Message | None' = None, *, before_turn: 'Callable[[int], Awaitable[None]] | None' = None, before_tools: 'Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]] | None' = None, providers: 'list[ToolProvider]' = <factory>, instructions: 'list[Instruction]' = <factory>, required_instructions: 'tuple[str, ...]' = (), before_turn_hooks: 'tuple[Callable[[int], Awaitable[None]], ...]' = (), before_tools_hooks: 'tuple[Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]], ...]' = (), system_contributions: 'tuple[Callable[[], str], ...]' = (), tool_guards: 'tuple[PermissionGuard, ...]' = ())

Bases: object

before_tools: Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]] | None

before_tools_hooks: tuple[Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]], ...]

before_turn: Callable[[int], Awaitable[None]] | None

before_turn_hooks: tuple[Callable[[int], Awaitable[None]], ...]

copy(**overrides: Any) → Self

Return a new Agent with fresh instruction state and overrides applied.

async dispatch_tools(blocks: list[ToolUseBlock], iteration: int) → list[ToolResultBlock]

get_tool(name: str) → Tool[Any] | None

Return the dispatchable tool, including its access declaration and guards.

During a run this uses the current iteration’s selected tool set. Preflight hooks can inspect effects without accessing a second registry.

get_tools() → list[Tool[Any]]

Return collected tools before selection, including activated instructions.

instructions: list[Instruction]

last_iteration_message: Message | None

async load_instruction(id: str) → LoadedInstruction

Load provider content and stage it for the next model iteration.

max_iterations: int

providers: list[ToolProvider]

required_instructions: tuple[str, ...]

async run(user_message: str, context: ContextStore) → str

run_stream(user_message: str, context: ContextStore) → AgentStream

selector: ToolSelector | None

system: str

system_contributions: tuple[Callable[[], str], ...]

tool_guards: tuple[PermissionGuard, ...]

tools: list[Tool[Any]]

transport: CompletionTransport

class axio.AgentStream(generator: AsyncGenerator[StreamEvent, None])

Bases: object

async aclose() → None

async get_final_text() → str

Everything the turn said, and nothing about whether it finished saying it.

A run ending on one of INCOMPLETE returns a truncated answer that reads exactly like a whole one, because a str has nowhere to put the reason. It is logged as a warning here, and get_session_end() carries it for a caller that needs to branch on it. Error still raises, so a broken turn is never returned as an answer.

async get_session_end() → SessionEndEvent

class axio.AudioOutput(index: int, data: bytes, media_type: AudioMediaType)

Bases: object

Audio content from a tool result (e.g. read_file on an audio file).

data: bytes

index: int

media_type: AudioMediaType

class axio.AudioOutputDelta(data: bytes, media_type: str = 'audio/pcm;rate=24000')

Bases: object

Streaming audio chunk from the assistant in a realtime session.

data: bytes

media_type: str

class axio.BlockEnd(index: int)

Bases: object

The content block at index is complete.

The point at which accumulated ToolInputDelta fragments are guaranteed to parse. Every provider marks it and axio has had no terminator for it until now.

index: int

class axio.Citation(index: int, cited_text: str = '', title: str | None = None, url: str | None = None, source_id: str | None = None, start: int | None = None, end: int | None = None, unit: Literal['char', 'byte', 'page', 'block', 'unknown']='unknown', raw: dict[str, ~typing.Any]=<factory>)

Bases: object

A span of generated text attributed to a source.

What Anthropic’s citation shapes, OpenAI’s annotations and Google’s grounding metadata have in common. raw keeps the provider’s whole object for the fields that differ.

cited_text: str

end: int | None

index: int

raw: dict[str, Any]

source_id: str | None

Whatever identifies the source inside this request: a file id, a document index, a chunk id.

start: int | None

title: str | None

unit: Literal['char', 'byte', 'page', 'block', 'unknown']

What start and end count. Stated because the providers disagree — OpenAI counts characters and Google counts bytes — so offsets from different units must never be compared.

url: str | None

class axio.CompletionTransport(*args, **kwargs)

Bases: Protocol

stream(messages: list[Message], tools: list[Tool[Any]], system: str) → AsyncIterator[StreamEvent]

class axio.ConcurrentGuard

Bases: PermissionGuard, ABC

Guard with concurrency control.

Subclass and override check(). __call__ acquires the semaphore then delegates to check(). Set concurrency to control parallelism (default 1 - one check at a time).

concurrency: int = 1

class axio.ContextStore

Bases: ABC

async add_context_tokens(input_tokens: int, output_tokens: int) → None

abstractmethod async append(message: Message) → None

async clear() → None

async close() → None

No-op by default.

async fork() → ContextStore

Returns a MemoryContextStore deep copy by default.

async classmethod from_context(context: ContextStore) → Self

async classmethod from_history(history: list[Message]) → Self

Create a new ContextStore pre-populated with history.

async get_context_tokens() → tuple[int, int]

Returns (0, 0) by default.

abstractmethod async get_history() → list[Message]

async list_sessions() → list[SessionInfo]

List available sessions. Default: returns a single entry for the current session.

property session_id: str

Lazy-init UUID hex; works without calling super().__init__().

async set_context_tokens(input_tokens: int, output_tokens: int) → None

No-op by default; tokens are silently dropped.

class axio.Error(exception: 'BaseException')

Bases: object

exception: BaseException

axio.Field(description: str = '', default: Any = MISSING, ge: int | float | None = None, le: int | float | None = None) → FieldInfo

Annotate a handler parameter with metadata (description, default, constraints).

Usage:

async def search(
    query: Annotated[str, Field(description="Search query")],
    limit: Annotated[int, Field(default=10, ge=1, le=100)],
) -> str: ...

class axio.FieldInfo(description: str = '', default: Any = MISSING, ge: int | float | None = None, le: int | float | None = None, strict: bool = False)

Bases: object

Metadata attached to a handler parameter via Annotated[T, FieldInfo(...)].

default: Any = MISSING

description: str = ''

ge: int | float | None = None

le: int | float | None = None

strict: bool = False

validate(value: Any, name: str, hint: Any) → None

Validate value against this FieldInfo’s constraints, raising if invalid.

exception axio.GuardError

Bases: ToolError

Guard denied or crashed during permission check.

exception axio.HandlerError

Bases: ToolError

Handler raised during execution.

class axio.ImageOutput(index: int, data: bytes, media_type: ImageMediaType)

Bases: object

Model generated an image inline (e.g. Nano Banana / Gemini Image).

data: bytes

index: int

media_type: ImageMediaType

class axio.IterationEnd(iteration: 'int', stop_reason: 'StopReason', usage: 'Usage', raw: 'str' = '')

Bases: object

iteration: int

raw: str

The provider’s own word for why it stopped, where it differs from what axio calls it. Kept so a caller can act on a reason this vocabulary has no name for.

stop_reason: StopReason

usage: Usage

class axio.IterationStart(iteration: int, id: str | None = None, model: str | None = None)

Bases: object

One provider request has begun.

model is the model that actually served the turn, which need not be the one asked for. Server-side fallback, sticky routing and dated-snapshot resolution all substitute a different model at a different price. A cost lookup therefore keys off this rather than off the request.

id: str | None

iteration: int

model: str | None

class axio.MemoryContextStore(history: list[Message] | None = None)

Bases: ContextStore

Simple in-memory context store. fork() returns a deep copy.

async append(message: Message) → None

async clear() → None

async close() → None

No-op by default.

async fork() → MemoryContextStore

Returns a MemoryContextStore deep copy by default.

async get_context_tokens() → tuple[int, int]

Returns (0, 0) by default.

async get_history() → list[Message]

property session_id: str

Lazy-init UUID hex; works without calling super().__init__().

async set_context_tokens(input_tokens: int, output_tokens: int) → None

No-op by default; tokens are silently dropped.

class axio.Message(role: "Literal['user', 'assistant', 'system']", content: 'list[ContentBlock]' = <factory>)

Bases: object

content: list[ContentBlock]

classmethod from_dict(data: dict[str, Any]) → Message

role: Literal['user', 'assistant', 'system']

to_dict() → dict[str, Any]

class axio.PermissionGuard

Bases: ABC

Gate for tool calls. Return modified kwargs to allow, raise to deny.

Tool invokes guards via await guard(tool, **kwargs).

Guards receive the Tool object and the raw keyword arguments before execution. Return the (possibly modified) dict to allow; raise GuardError to deny. Because guards see all inputs up front, they are also the right place for logging and auditing:

class AuditGuard(PermissionGuard):
    async def check(self, tool: Tool, **kwargs: Any) -> dict[str, Any]:
        logger.info("tool=%s args=%s", tool.name, kwargs)
        return kwargs  # always allow

See examples/agent_swarm/agent_swarm/__main__.py (RoleGuard) for a production example.

abstractmethod async check(tool: Tool[Any], **kwargs: Any) → dict[str, Any]

class axio.ProviderBlock(provider: str, kind: str, data: dict[str, Any], id: str = '')

Bases: ContentBlock

One item of the turn that only the protocol which produced it can read.

An endpoint that runs its own tools answers with items this vocabulary has no shape for: a web search it ran, a file it read, code it executed, and whatever it adds next. Where the application keeps the history rather than the provider — store=False on the Responses API — the provider expects every one of those items back, in order and exactly once. Dropped, the next request is missing what the model was answering from, and reasoning continuity, tool-call association and anything built on a newer item type break with nothing said.

data is the item exactly as it arrived and is never interpreted here. provider names the protocol that issued it, so it is replayed only to an endpoint that speaks the same one, and kind is the item’s own type, which is all a caller needs to render or count it.

data: dict[str, Any]

id: str

How the provider names this item, where it names them.

kind: str

provider: str

class axio.ProviderEvent(provider: str, kind: str, data: dict[str, Any], index: int | None = None)

Bases: object

A provider payload axio does not model, forwarded verbatim.

data is the provider’s own JSON object exactly as it was parsed: no renaming, no coercion, no filtering.

How completely a transport forwards depends on what its stream names. A stream that names each event has a reader. That reader forwards every payload it does not interpret, so nothing is dropped. A stream with no discriminator — one shape per payload, read field by field — has no such catch. Those transports forward the parts they know how to name. A field the provider adds inside a payload reaches nobody until someone reads it.

A consumer that does not recognise (provider, kind) ignores it.

data: dict[str, Any]

index: int | None

The content-block or output index, where the payload carries one.

kind: str

The provider’s own discriminator, verbatim, never a name axio invented: matching on it is matching the vocabulary the provider publishes.

provider: str

Which transport produced it: "anthropic", "openai", "google". The Codex transport reads its stream through the shared Responses reader, so its events say "openai" rather than naming a fourth provider.

class axio.ProviderOutput(provider: str, kind: str, data: dict[str, Any], index: int, id: str = '')

Bases: object

One output item of this turn that the next request has to send back unread.

ProviderEvent forwards what a caller may want to watch; this one names what the turn is not complete without. An endpoint that keeps no history of its own — store=False on the Responses API — expects every item it produced back on the next request, including those from the tools it ran itself: a web search, a file search, a code interpreter, and item types that do not exist yet. Read as news rather than as content, they were watched and dropped, and the next request was missing what the model had answered from.

The agent stores it as a ProviderBlock, and the transport that speaks the same protocol replays data verbatim.

data: dict[str, Any]

The item exactly as it arrived, never interpreted here.

id: str

index: int

The output index this item occupied. Declared in the same position as on ProviderEvent, whose first three fields these are: with the two orders crossed, one built positionally in the shape of the other assigned every field to the wrong name.

kind: str

The item’s own type, verbatim.

provider: str

Which protocol produced it, in the names ProviderEvent uses.

class axio.RealtimeAgent(system: str, transport: RealtimeTransport, tools: list[Tool[Any]] = <factory>, voice: str | None = None, input_audio_format: str = 'audio/pcm;rate=16000', output_audio_format: str = 'audio/pcm;rate=24000', raise_on_error: bool = True)

Bases: object

Drives a duplex realtime session against a RealtimeTransport.

The agent intercepts ToolUseStart/ToolInputDelta/TurnComplete events to assemble tool calls and dispatches them as background tasks so that streaming audio output from the provider is not blocked. Each tool result is sent back to the session via RealtimeSession.send_tool_result() as soon as the handler returns. interrupt() cancels in-flight tasks.

Lifecycle:

async with RealtimeAgent(system="...", transport=t, tools=[...]) as agent:
    await agent.send(AudioBlock(data=mic_chunk, media_type="audio/pcm"))
    async for event in agent.events():
        match event:
            case AudioOutputDelta(data=pcm):
                speaker.feed(pcm)
            ...

async close() → None

async commit() → None

async connect() → None

async events() → AsyncIterator[StreamEvent]

Yield events from the session, dispatching tool calls as side-effects.

Tool dispatch runs as background tasks — slow handlers do not block the stream of AudioOutputDelta from the provider.

input_audio_format: str

async interrupt() → None

output_audio_format: str

raise_on_error: bool

Re-raise the exception wrapped in any Error event yielded by the session. Set to False to receive Error events as data and decide what to do per error (transient retry, log-and-continue, etc.).

async send(content: ContentBlock | list[ContentBlock]) → None

property session: RealtimeSession

system: str

tools: list[Tool[Any]]

transport: RealtimeTransport

voice: str | None

class axio.RealtimeSession(*args, **kwargs)

Bases: Protocol

Active duplex realtime session — bidirectional audio / text / tools.

Returned by RealtimeTransport.connect(). Events from the provider arrive on events(). User input is pushed via send().

async close() → None

Tear down the session and release resources.

async commit() → None

Signal end-of-utterance for manual VAD; no-op with server VAD.

events() → AsyncIterator[StreamEvent]

Async iterator over server events for the lifetime of this session.

async interrupt() → None

Abort in-flight assistant generation (e.g. user interrupted).

async send(content: ContentBlock | list[ContentBlock]) → None

Append user content (audio chunk, text, image) to the input buffer.

async send_tool_result(tool_use_id: ToolCallID, name: ToolName, content: str | list[ContentBlock]) → None

Deliver a tool’s result to the provider so generation can resume.

name is included because some providers (e.g. Gemini Live) require the tool name alongside the call id. OpenAI realtime can ignore it.

class axio.RealtimeTransport(*args, **kwargs)

Bases: Protocol

Provider that supports duplex realtime sessions (e.g. OpenAI Realtime, Gemini Live). Distinct from CompletionTransport because the interaction is bidirectional, not request/response.

async connect(*, system: str, tools: list[Tool[Any]], voice: str | None = None, input_audio_format: str = 'audio/pcm;rate=16000', output_audio_format: str = 'audio/pcm;rate=24000') → RealtimeSession

class axio.ReasoningBlock(text: str = '', signature: str = '', redacted: bool = False, id: str = '', provider: str = '')

Bases: ContentBlock

The model’s own reasoning, kept so the turn can be sent back unaltered.

signature is the provider’s proof that the block is its own. Anthropic refuses a returned thinking block whose signature is missing or changed, and Google reports MISSING_THOUGHT_SIGNATURE for the same failure. A stored turn that dropped the signature cannot be replayed. Never inspect, decode or truncate it.

redacted marks a block whose reasoning the provider withheld. The signature still has to travel, and text is empty.

id: str

How the provider names this block, where it names them. Required beside the signature.

provider: str

Which protocol issued signature: "anthropic", "google", "openai", the same names ProviderEvent uses. The value means nothing outside the protocol that made it — Anthropic sends it as a thinking signature, Google as thoughtSignature, Responses as encrypted_content — so a session that changes transport must not replay it to the next one. Empty says nobody recorded it, which is what a turn stored before this field existed looks like; proof() lets that through rather than dropping proofs it cannot judge.

redacted: bool

signature: str

Out of repr on purpose: this is the one field documented as never to be inspected, and a debug log of the block printed it in full beside everything else.

text: str

class axio.ReasoningDelta(index: 'int', delta: 'str')

Bases: object

delta: str

index: int

class axio.ReasoningSignature(index: int, signature: str, redacted: bool = False, id: str = '', provider: str = '')

Bases: object

Opaque proof that a reasoning block is the provider’s own, to be replayed unaltered.

Anthropic refuses a returned thinking block whose signature is missing or changed, and Google publishes a MISSING_THOUGHT_SIGNATURE finish reason for the same failure. Never inspect, decode, re-encode or truncate data.

id: str

How the provider names the block this proves, where it names them. Replayed beside the proof, because a provider that identifies reasoning by id refuses the pair without it.

index: int

provider: str

Which protocol issued this proof, in the names ProviderEvent uses.

redacted: bool

True where the payload replaces the reasoning text instead of accompanying it.

signature: str

class axio.Refusal(index: int, text: str = '', spoken: bool = True, category: str | None = None, blocked_input: bool = False, raw: dict[str, ~typing.Any]=<factory>)

Bases: object

The model declined, or the provider blocked the turn.

This is deliberately not a TextDelta. As ordinary assistant text, or as an empty turn that succeeded, a refusal is indistinguishable from an answer and no consumer can act on it.

blocked_input: bool

True where the provider rejected the prompt rather than the answer, so nothing was generated and sending the same prompt again cannot succeed.

category: str | None

The provider’s own category, verbatim. Not normalised: the taxonomies do not overlap, and a mapping between them would state something no provider says.

index: int

raw: dict[str, Any]

spoken: bool

Whether text is the model’s own words.

True on the endpoints that stream a refusal as output content, which is what OpenAI and the Responses API do. False where the text is the provider explaining why it stopped: Anthropic sends stop_details.explanation, which its own schema documents as unstable and not to be parsed, and the model generated nothing at all.

The agent stores either kind as the turn’s text, because a stored turn with no content is refused by the next request and the explanation is the only account of the decline there is. A consumer that renders the two differently — the model declining, against the provider reporting a block — reads this.

text: str

class axio.SessionEndEvent(stop_reason: 'StopReason', total_usage: 'Usage')

Bases: object

stop_reason: StopReason

total_usage: Usage

class axio.SpeechStarted

Bases: object

Server VAD detected the user started speaking (realtime).

class axio.SpeechStopped

Bases: object

Server VAD detected the user stopped speaking (realtime).

class axio.StopReason(*values)

Bases: StrEnum

Why the provider stopped generating.

Anything that is not tool_use or pause_turn ends the run. See the match in Agent._run_loop, whose wildcard keeps a member added here from falling through into another paid iteration.

cancelled = 'cancelled'

The caller or the provider stopped the turn before it finished.

context_window_exceeded = 'context_window_exceeded'

The conversation outgrew the model’s window. Truncated, like max_tokens.

end_turn = 'end_turn'

error = 'error'

max_tokens = 'max_tokens'

pause_turn = 'pause_turn'

A server-side tool loop reached its iteration limit. Resumable: the provider expects the assistant content back so it can finish. This is the one reason that does not end the run.

refusal = 'refusal'

The model declined, or the provider blocked the turn. Not an error: the same prompt sent again will be declined again.

repetition = 'repetition'

Axio stopped the turn, because the model was repeating itself. The only reason here the provider did not give. Reported as end_turn, a caller could not tell an answer the model finished from one cut off mid-word, which is the same objection this vocabulary raises against reading a truncated response as a whole one.

tool_use = 'tool_use'

unknown = 'unknown'

The provider said something this vocabulary does not have. Terminal, and it vouches for nothing. Named rather than folded into one of the others, because each of those claims something the provider did not say: that the turn finished, that it was truncated, or that the transport broke. IterationEnd.raw carries the word itself.

type axio.StreamEvent = ReasoningDelta | ReasoningSignature | TextDelta | TextSignature | Refusal | Citation | ImageOutput | AudioOutput | VideoOutput | ToolUseStart | ToolInputDelta | ToolFieldStart | ToolFieldDelta | ToolFieldEnd | ToolOutputDelta | ToolResult | BlockEnd | IterationStart | IterationEnd | Error | ProviderEvent | ProviderOutput | SessionEndEvent | AudioOutputDelta | TranscriptDelta | SpeechStarted | SpeechStopped | TurnComplete

class axio.TextBlock(text: 'str', signature: 'str' = '', provider: 'str' = '')

Bases: ContentBlock

provider: str

Which protocol issued signature. See ReasoningBlock.provider.

signature: str

Opaque proof for this text, where the provider signs a text part. Replayed with it. Kept out of repr with the rest of them: see provider below.

text: str

class axio.TextDelta(index: 'int', delta: 'str')

Bases: object

delta: str

index: int

class axio.TextSignature(index: int, signature: str, provider: str = '')

Bases: object

Opaque proof that a block of answer text is the provider’s own, to be replayed unaltered.

Google signs the part it issued the proof for, and answer text is one such part. The proof belongs to the text block, not to the reasoning or the call beside it: replayed on another part it proves nothing, and the turn fails with MISSING_THOUGHT_SIGNATURE. Never inspect, decode, re-encode or truncate data.

Emitted after the text it signs, never before.

index: int

provider: str

Which protocol issued this proof, in the names ProviderEvent uses.

signature: str

class axio.Tool(name: 'ToolName', handler: 'Callable[..., Awaitable[Any]]', description: 'str' = '', guards: 'tuple[PermissionGuard, ...]' = (), concurrency: 'int | None' = None, context: 'T' = mappingproxy({}), schema: 'MappingProxyType[str, Any]' = mappingproxy({}), *, access: 'AccessDescriptor | None' = None, permission_identity: 'str' = '')

Bases: Generic

access: AccessDescriptor | None

async call_streaming(**kwargs: Any) → AsyncGenerator[tuple[str, str], None]

Execute handler, yielding (key, text) chunks for streaming output.

Uses handler.stream(**kwargs) if the handler exposes one. Otherwise falls back to __call__() and yields the full result as a single ("output", ...) chunk. Semaphore is held for the entire iteration.

concurrency: int | None

context: T

description: str

format_stream_result(chunks: list[tuple[float, str, str]]) → str

Aggregate streamed chunks into the final tool result string.

Handlers may attach a format_stream_result callable for structured output (e.g. shell log records). Defaults to text concatenation.

guards: tuple[PermissionGuard, ...]

handler: Callable[[...], Awaitable[Any]]

property input_schema: JSONSchema

name: ToolName

permission_identity: str

schema: MappingProxyType

property supports_streaming: bool

Handler supports streaming if it exposes a .stream async-generator attribute.

class axio.ToolFieldDelta(index: 'int', tool_use_id: 'ToolCallID', key: 'str', text: 'str')

Bases: object

index: int

key: str

text: str

tool_use_id: ToolCallID

class axio.ToolFieldEnd(index: 'int', tool_use_id: 'ToolCallID', key: 'str')

Bases: object

index: int

key: str

tool_use_id: ToolCallID

class axio.ToolFieldStart(index: 'int', tool_use_id: 'ToolCallID', key: 'str')

Bases: object

index: int

key: str

tool_use_id: ToolCallID

class axio.ToolInputDelta(index: 'int', tool_use_id: 'ToolCallID', partial_json: 'str')

Bases: object

index: int

partial_json: str

tool_use_id: ToolCallID

class axio.ToolOutputDelta(tool_use_id: 'ToolCallID', name: 'ToolName', key: 'str', delta: 'str')

Bases: object

delta: str

key: str

name: ToolName

tool_use_id: ToolCallID

class axio.ToolResult(tool_use_id: 'ToolCallID', name: 'ToolName', is_error: 'bool', content: 'str' = '', input: 'dict[str, Any]'=<factory>)

Bases: object

content: str

input: dict[str, Any]

is_error: bool

name: ToolName

tool_use_id: ToolCallID

class axio.ToolResultBlock(tool_use_id: 'ToolCallID', content: 'str | list[TextBlock | ImageBlock | AudioBlock | VideoBlock]', is_error: 'bool' = False)

Bases: ContentBlock

content: str | list[TextBlock | ImageBlock | AudioBlock | VideoBlock]

is_error: bool

tool_use_id: ToolCallID

class axio.ToolSelector(*args, **kwargs)

Bases: Protocol

async select(messages: Iterable[Message], tools: Iterable[Tool[Any]]) → Iterable[Tool[Any]]

class axio.ToolUseBlock(id: 'ToolCallID', name: 'ToolName', input: 'dict[str, Any]', signature: 'str' = '', provider: 'str' = '')

Bases: ContentBlock

id: ToolCallID

input: dict[str, Any]

name: ToolName

provider: str

Which protocol issued signature. See ReasoningBlock.provider.

signature: str

Opaque proof for this call. Stored rather than held in the transport, which a restart empties.

class axio.ToolUseStart(index: 'int', tool_use_id: 'ToolCallID', name: 'ToolName', signature: 'str' = '', provider: 'str' = '')

Bases: object

index: int

name: ToolName

provider: str

Which protocol issued signature, in the names ProviderEvent uses. Stored with the proof so a session that changes transport does not replay it to one that never made it.

signature: str

Opaque proof that this call is the model’s own, where the provider issues one for the call rather than for the reasoning beside it. Replayed on the call itself: attached to a reasoning block instead, the call comes back unsigned and the provider refuses the turn.

tool_use_id: ToolCallID

class axio.TranscriptDelta(role: Literal['user', 'assistant'], delta: str)

Bases: object

Live transcript delta — server-side STT of user mic, or assistant speech transcription, depending on role.

delta: str

role: Literal['user', 'assistant']

class axio.TurnComplete(stop_reason: StopReason, usage: Usage | None = None)

Bases: object

Assistant turn finished in a realtime session. stop_reason may be StopReason.tool_use to signal that pending tool calls should run before the next turn starts.

stop_reason: StopReason

usage: Usage | None

class axio.Usage(input_tokens: int, output_tokens: int, *, cache_read_tokens: int = 0, cache_write_tokens: int = 0, reasoning_tokens: int = 0)

Bases: object

Token counts for one provider request.

The rule: input_tokens and output_tokens are always inclusive grand totals, and every other field is a disjoint slice of one of them:

cache_read_tokens + cache_write_tokens  <=  input_tokens
reasoning_tokens                        <=  output_tokens

Providers disagree about whether their own headline number already contains the slices, and they disagree in opposite directions. Anthropic counts only the tokens after the last cache breakpoint, so its cache counts have to be added. Google reports thinking beside the candidates rather than inside them. Each transport adds or does not add to satisfy the rule here, so nothing downstream has to know which provider answered.

Counts only, never money. A cached token and a written one bill at different multipliers, so a caller that wants cost multiplies these slices by its own per-model rates. A zero slice means the provider billed none of it, or reported no breakdown at all. Axio cannot tell those apart.

property answer_tokens: int

Output that was the answer rather than reasoning.

cache_read_tokens: int

The slice of input_tokens served from cache, billed at a discount.

cache_write_tokens: int

The slice of input_tokens written to cache, billed at a premium. Disjoint from the read.

input_tokens: int

output_tokens: int

reasoning_tokens: int

The slice of output_tokens spent on reasoning the caller never sees.

classmethod reported(input_tokens: int, output_tokens: int, *, cache_read_tokens: int = 0, cache_write_tokens: int = 0, reasoning_tokens: int = 0) → Usage

What a provider said, held to the rule above rather than trusted to follow it.

Constructed directly, a provider that reports a slice larger than the total it belongs to fails __post_init__, and a whole answer is lost over an accounting discrepancy. Every transport reads its provider’s numbers through here instead.

A slice that outgrew its total means the total was reported without it. The total is therefore raised to hold the slices, which keeps the tokens the provider billed for; the other repair, cutting the slice down, throws them away and under-reports the cost.

property total_tokens: int

property uncached_input_tokens: int

Input the provider had to read in full, which is what most of the bill is.

class axio.VideoOutput(index: int, data: bytes, media_type: VideoMediaType)

Bases: object

Model generated a video inline.

data: bytes

index: int

media_type: VideoMediaType

Agent: the core agentic loop orchestrating transport, tools, and context.

class axio.agent.Agent(system: 'str', transport: 'CompletionTransport', tools: 'list[Tool[Any]]' = <factory>, selector: 'ToolSelector | None' = None, max_iterations: 'int' = 50, last_iteration_message: 'Message | None' = None, *, before_turn: 'Callable[[int], Awaitable[None]] | None' = None, before_tools: 'Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]] | None' = None, providers: 'list[ToolProvider]' = <factory>, instructions: 'list[Instruction]' = <factory>, required_instructions: 'tuple[str, ...]' = (), before_turn_hooks: 'tuple[Callable[[int], Awaitable[None]], ...]' = (), before_tools_hooks: 'tuple[Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]], ...]' = (), system_contributions: 'tuple[Callable[[], str], ...]' = (), tool_guards: 'tuple[PermissionGuard, ...]' = ())

Bases: object

before_tools: Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]] | None

before_tools_hooks: tuple[Callable[[list[ToolUseBlock], int], Awaitable[list[ToolResultBlock]]], ...]

before_turn: Callable[[int], Awaitable[None]] | None

before_turn_hooks: tuple[Callable[[int], Awaitable[None]], ...]

copy(**overrides: Any) → Self

Return a new Agent with fresh instruction state and overrides applied.

async dispatch_tools(blocks: list[ToolUseBlock], iteration: int) → list[ToolResultBlock]

get_tool(name: str) → Tool[Any] | None

Return the dispatchable tool, including its access declaration and guards.

During a run this uses the current iteration’s selected tool set. Preflight hooks can inspect effects without accessing a second registry.

get_tools() → list[Tool[Any]]

Return collected tools before selection, including activated instructions.

instructions: list[Instruction]

last_iteration_message: Message | None

async load_instruction(id: str) → LoadedInstruction

Load provider content and stage it for the next model iteration.

max_iterations: int

providers: list[ToolProvider]

required_instructions: tuple[str, ...]

async run(user_message: str, context: ContextStore) → str

run_stream(user_message: str, context: ContextStore) → AgentStream

selector: ToolSelector | None

system: str

system_contributions: tuple[Callable[[], str], ...]

tool_guards: tuple[PermissionGuard, ...]

tools: list[Tool[Any]]

transport: CompletionTransport

type axio.agent.TurnBlock = TextBlock | ReasoningBlock | ImageBlock | AudioBlock | VideoBlock | ToolUseBlock | ProviderBlock

What one iteration accumulates while the transport streams it.

Tool: frozen dataclass binding a handler callable to a name, guards, and concurrency.

type axio.tool.JSONSchema = dict[str, Any]

class axio.tool.Tool(name: 'ToolName', handler: 'Callable[..., Awaitable[Any]]', description: 'str' = '', guards: 'tuple[PermissionGuard, ...]' = (), concurrency: 'int | None' = None, context: 'T' = mappingproxy({}), schema: 'MappingProxyType[str, Any]' = mappingproxy({}), *, access: 'AccessDescriptor | None' = None, permission_identity: 'str' = '')

Bases: Generic

access: AccessDescriptor | None

async call_streaming(**kwargs: Any) → AsyncGenerator[tuple[str, str], None]

Execute handler, yielding (key, text) chunks for streaming output.

Uses handler.stream(**kwargs) if the handler exposes one. Otherwise falls back to __call__() and yields the full result as a single ("output", ...) chunk. Semaphore is held for the entire iteration.

concurrency: int | None

context: T

description: str

format_stream_result(chunks: list[tuple[float, str, str]]) → str

Aggregate streamed chunks into the final tool result string.

Handlers may attach a format_stream_result callable for structured output (e.g. shell log records). Defaults to text concatenation.

guards: tuple[PermissionGuard, ...]

handler: Callable[[...], Awaitable[Any]]

property input_schema: JSONSchema

name: ToolName

permission_identity: str

schema: MappingProxyType

property supports_streaming: bool

Handler supports streaming if it exposes a .stream async-generator attribute.

axio.tool.hint_from_json_schema(prop_schema: dict[str, Any]) → Any

Return the Python type hint for a single JSON Schema property definition.

Instructions

axio.instructions.Instruction is an abstract base class for instructions in general: rules, reference material, procedures, and skills. A skill is one use of this contract. Core does not interpret skill formats or provider locations.

A provider supplies an identifier, a short summary, an optional source label, and an asynchronous load() implementation. Reading the metadata does not load the content. Use an identifier that distinguishes sources with the same name.

from __future__ import annotations

import asyncio

from axio.instructions import Instruction, LoadedInstruction
from axio.tool import Tool


async def convert_temperature(celsius: float) -> str:
    """Convert degrees Celsius to degrees Fahrenheit."""
    return str(celsius * 9 / 5 + 32)


class TemperatureInstructions(Instruction):
    async def load(self) -> LoadedInstruction:
        return LoadedInstruction(
            "Use convert_temperature and include the unit in the answer.",
            (Tool(name="convert_temperature", handler=convert_temperature),),
        )


async def example() -> None:
    instruction = TemperatureInstructions("example:temperature", "Temperature conversion")
    assert instruction.summary == "Temperature conversion"
    content = await instruction.load()
    assert await content.tools[0](celsius=0.0) == "32.0"


asyncio.run(example())

Instructions without tools return LoadedInstruction(text). Each call to load() requests current content; providers may cache it only while they can validate its freshness. Loading does not execute the described procedure or change an agent. The caller decides when to load mandatory or optional content.

Exceptions and cancellation propagate to the caller. A provider cleans up resources acquired by a failed load and owns sessions used by returned tools. Those sessions must remain open while the tools are in use. Tools keep their guards and access declarations; loading an instruction does not grant access.

Use instructions with an agent

Pass instruction objects in Agent(instructions=[...]). The agent adds their short summaries and sources to the system prompt and exposes load_instruction. The model calls this tool with an instruction ID to read the full content. Associated tools become available on the next model iteration. They cannot run in the same batch that loads them.

For instructions that must apply before the first model request, pass their IDs in required_instructions. These instructions are refreshed before each model iteration. Providers should reuse their tool objects when the tools are unchanged. Conflicting instruction IDs or tool names raise an error; they never silently replace another provider’s declaration.

from __future__ import annotations

import asyncio

from axio import Agent, MemoryContextStore
from axio.instructions import Instruction, LoadedInstruction
from axio.testing import StubTransport, make_text_response, make_tool_use_response


class StyleRules(Instruction):
    async def load(self) -> LoadedInstruction:
        return LoadedInstruction("Use short sentences and state assumptions explicitly.")


async def example() -> None:
    rules = StyleRules("example:style", "Writing conventions")
    transport = StubTransport([
        make_tool_use_response("load_instruction", tool_input={"id": rules.id}),
        make_text_response("The rules are loaded."),
    ])
    agent = Agent("Answer the user.", transport, instructions=[rules])
    assert await agent.run("Read the writing conventions.", MemoryContextStore()) == "The rules are loaded."


asyncio.run(example())

The agent collects base tools and loaded instruction tools, applies tool_guards, and then calls its ToolSelector. Only tools in that iteration’s selected set can execute. Application guards supplement the guards supplied by each tool. They also apply to tools added by instructions and to load_instruction itself.

before_turn_hooks and before_tools_hooks run in tuple order after the existing before_turn and before_tools callbacks. Tool hooks receive the calls that earlier hooks did not defer. Each hook must return results with distinct IDs from those calls. A failed hook prevents the whole batch from executing.

A preflight hook can call await agent.load_instruction(id) to load newly applicable mandatory instructions. If content changes, the whole batch is deferred so the model can reconsider it with the updated instructions. Unchanged content and identical tool objects do not trigger another deferral. Hook errors and cancelled runs discard staged changes.

system_contributions contains synchronous callbacks that supply additional prompt sections. The agent combines the base system, these sections, and its current instruction content without changing the base string.

Replace agent.instructions between runs to reload a catalog. Removed or replaced objects lose their active content and associated tools at the next iteration. agent.copy() creates independent instruction state and a separate catalog list. The caller must keep each provider open while any agent uses its tools. Pass initialized providers with Agent(..., providers=[provider]). The agent reads their catalogs at each iteration; applications own initialization, reload, and cleanup. agent.get_tools() returns currently available tools before selection, including activated tools and application guards.

Instruction.load_access() can declare the effects of loading content for permission guards. The default is unknown effects. Local skills declare their source file as a read. These declarations do not grant permission. Required instructions are loaded by the application-controlled preflight path.

Provider-independent instructions with content loaded on demand.

class axio.instructions.Instruction(id: str, summary: str, *, source: str = '')

Bases: ABC

An instruction whose provider implements asynchronous content loading.

id identifies the instruction within a catalog. Providers must include enough source identity to distinguish instructions with the same name. summary is the short description available before loading the content. source is an optional display location; core does not interpret it.

Rules, reference material, procedures, and skills use the same contract. The caller decides when loading is required. This class does not infer applicability from a filename, URI, or instruction type.

abstractmethod async load() → LoadedInstruction

Return complete content and any tools needed to follow it.

Implementations prepare dependencies before returning. They must not change an agent or execute the procedure described by the instruction. Exceptions, including cancellation, propagate to the caller. Providers must clean up resources acquired by a failed or cancelled load.

Each call requests the current content. A provider may use a cache if it can validate its freshness. Core does not cache this method or share activation state between agents. Providers retain ownership of any sessions used by the returned tools.

load_access() → tuple[Access, ...]

Declare loading effects for application guards; unknown unless overridden.

class axio.instructions.InstructionState

Bases: object

One agent’s catalog, loaded content, and changes awaiting an iteration.

Providers retain their resources. This object only owns the agent’s view of their instructions. A successful load stages changes; commit makes them available together at the next model boundary.

static collect_tools(base: Iterable[Tool[Any]], content: Iterable[LoadedInstruction]) → list[Tool[Any]]

Preserve tool identities and reject conflicting names before activation.

commit(base: Iterable[Tool[Any]]) → list[Tool[Any]]

Apply staged content atomically and return the current tool set.

async load(id: str, base: Iterable[Tool[Any]]) → LoadedInstruction

Stage one complete provider result; failure leaves the state unchanged.

render() → str

Render the short catalog and the content currently active for this agent.

synchronize(instructions: Iterable[Instruction]) → None

Replace catalog metadata and discard removed or replaced content.

class axio.instructions.LoadedInstruction(text: str, tools: tuple[Tool[Any], ...] = ())

Bases: object

A complete instruction and the tools needed to follow it.

Tools retain their guards and access declarations. Their provider owns any resources they use and must keep those resources open while they are in use. Returning a tool does not execute it or grant permission to call it.

text: str

tools: tuple[Tool[Any], ...]

Provider discovery

See Register and discover plugins for all entry-point groups, installation and registration examples, lifecycle contracts, and discovery diagnostics.

axio.plugins loads installed entry points into PluginCatalog. Each catalog retains entries, load errors, and the declaring distribution. The first declaration reserves its name even when loading fails. require(name) reports a stored failure or lists the available names when the name is unknown.

The groups have separate contracts:

Group	Target
axio.tools	A plain async tool handler
axio.tools.providers	A zero-argument ToolProvider factory
axio.guards	A PermissionGuard subclass
axio.selector	A ToolSelector class
axio.context	A zero-argument ContextBackend factory
axio.transport	A transport class or plugin, through discover_transports()

A tool provider implements init(config=None, global_config=None), close(), the all_tools property, and get_instructions(). The last method returns Instruction objects with metadata available before loading. Provider factories must not acquire resources: initialization and cleanup belong to the lifecycle.

Register a provider in its package:

[project.entry-points."axio.tools.providers"]
example = "example_tools:Provider"

Use open_tool_providers() as an async context manager. It registers cleanup before each initialization and closes providers in reverse order, including on partial startup, cancellation, and exceptions in the context body. A startup failure propagates after cleanup. Use separate contexts to isolate independent startup failures. Providers must permit close() after partial initialization.

collect_tools() reads the current tool sets and rejects conflicting names. Multiple references to the same tool object are allowed. It does not cache a provider’s tools, so callers can collect them again after a provider reload.

standalone_tools() excludes plain handlers also supplied by an initialized provider from the same distribution. Tools from different distributions still conflict when their names match.

The TUI retains its settings groups and ToolsPlugin interface. Settings screens are separate from the core provider contract. Core imports no UI implementation.

Common entry-point discovery and tool-provider lifecycle, without UI imports.

class axio.plugins.ContextBackend(*args, **kwargs)

Bases: Protocol

An installed store factory that owns resources shared by its sessions.

async close() → None

create(session_id: str | None = None) → ContextStore

async init(path: Path) → None

class axio.plugins.PluginCatalog(entries: dict[str, T]=<factory>, errors: dict[str, str]=<factory>, sources: dict[str, PluginSource]=<factory>)

Bases: Generic

Successfully loaded extensions, failures, and their installed sources.

entries: dict[str, T]

errors: dict[str, str]

require(name: str) → T

sources: dict[str, PluginSource]

exception axio.plugins.PluginError

Bases: RuntimeError

An extension could not be loaded or conflicts with another extension.

class axio.plugins.PluginSource(group: str, distribution: str, value: str)

Bases: object

The installed declaration that owns an entry-point name.

distribution: str

group: str

value: str

class axio.plugins.RefreshableToolProvider(*args, **kwargs)

Bases: Protocol

An optional capability for refreshing tools and instructions in place.

async refresh(*, config: Any = None) → None

class axio.plugins.ToolLifecycle(*args, **kwargs)

Bases: Protocol

The working interface shared by current and legacy tool providers.

property all_tools: list[Tool[Any]]

async close() → None

async init(config: Any = None, global_config: Any = None) → None

class axio.plugins.ToolProvider(*args, **kwargs)

Bases: ToolLifecycle, Protocol

A dynamic tool set and instruction catalog owned by an installed provider.

Register a zero-argument factory in axio.tools.providers. The factory must not acquire resources. init receives application configuration; only the provider interprets its contents. close must also work after a failed or cancelled initialization.

get_instructions() → Sequence[Instruction]

axio.plugins.collect_tools(providers: Iterable[ToolLifecycle], *, tools: Iterable[Tool[Any]] = ()) → list[Tool[Any]]

Collect current tools, rejecting different tools with the same name.

axio.plugins.context_backend_from_target(name: str, target: Any) → ContextBackend

axio.plugins.discover_context_backends() → PluginCatalog[ContextBackend]

axio.plugins.discover_guards(*, points: Iterable[EntryPoint] | None = None) → PluginCatalog[type[PermissionGuard]]

axio.plugins.discover_selectors(*, points: Iterable[EntryPoint] | None = None) → PluginCatalog[type[ToolSelector]]

axio.plugins.discover_tool_providers(*, points: Iterable[EntryPoint] | None = None) → PluginCatalog[ToolProvider]

axio.plugins.discover_tools(*, points: Iterable[EntryPoint] | None = None) → PluginCatalog[Tool[Any]]

axio.plugins.guard_from_target(name: str, target: Any) → type[PermissionGuard]

axio.plugins.load_entry_points(group: str, adapt: Callable[[str, Any], T], *, points: Iterable[EntryPoint] | None = None) → PluginCatalog

Load each name once, retaining failures and reserving the first declaration.

Discovery does not initialize providers. An adapter validates the contract for its own group. A broken declaration cannot be replaced by a duplicate.

axio.plugins.open_tool_providers(providers: Iterable, *, config: Any = None, global_config: Any = None) → AsyncIterator[tuple[T, ...]]

Initialize providers and close them in reverse order on every exit path.

Cleanup is registered before initialization so partial startup is covered. A startup error propagates after closing the failing and earlier providers. The caller may use separate contexts when startup failures must be isolated.

axio.plugins.provider_from_target(name: str, target: Any) → ToolProvider

axio.plugins.selector_from_target(name: str, target: Any) → type[ToolSelector]

axio.plugins.standalone_tools(tools: PluginCatalog[Tool[Any]], providers: PluginCatalog[ToolProvider]) → list[Tool[Any]]

Prefer a distribution’s initialized provider over its duplicate plain handlers.

Different distributions still conflict. Plain handlers without a matching provider tool remain available.

axio.plugins.tool_from_handler(name: str, handler: Any) → Tool[Any]

Validate a plain async handler and preserve its concurrency setting.

Context and events

ContextStore: protocol for conversation history storage.

class axio.context.ContextStore

Bases: ABC

async add_context_tokens(input_tokens: int, output_tokens: int) → None

abstractmethod async append(message: Message) → None

async clear() → None

async close() → None

No-op by default.

async fork() → ContextStore

Returns a MemoryContextStore deep copy by default.

async classmethod from_context(context: ContextStore) → Self

async classmethod from_history(history: list[Message]) → Self

Create a new ContextStore pre-populated with history.

async get_context_tokens() → tuple[int, int]

Returns (0, 0) by default.

abstractmethod async get_history() → list[Message]

async list_sessions() → list[SessionInfo]

List available sessions. Default: returns a single entry for the current session.

property session_id: str

Lazy-init UUID hex; works without calling super().__init__().

async set_context_tokens(input_tokens: int, output_tokens: int) → None

No-op by default; tokens are silently dropped.

class axio.context.MemoryContextStore(history: list[Message] | None = None)

Bases: ContextStore

Simple in-memory context store. fork() returns a deep copy.

async append(message: Message) → None

async clear() → None

async close() → None

No-op by default.

async fork() → MemoryContextStore

Returns a MemoryContextStore deep copy by default.

async get_context_tokens() → tuple[int, int]

Returns (0, 0) by default.

async get_history() → list[Message]

property session_id: str

Lazy-init UUID hex; works without calling super().__init__().

async set_context_tokens(input_tokens: int, output_tokens: int) → None

No-op by default; tokens are silently dropped.

class axio.context.SessionInfo(session_id: 'str', message_count: 'int', preview: 'str', created_at: 'str', input_tokens: 'int' = 0, output_tokens: 'int' = 0)

Bases: object

created_at: str

input_tokens: int

message_count: int

output_tokens: int

preview: str

session_id: str

Stream events: all variants emitted by AgentStream.

class axio.events.AudioOutput(index: int, data: bytes, media_type: AudioMediaType)

Bases: object

Audio content from a tool result (e.g. read_file on an audio file).

data: bytes

index: int

media_type: AudioMediaType

class axio.events.AudioOutputDelta(data: bytes, media_type: str = 'audio/pcm;rate=24000')

Bases: object

Streaming audio chunk from the assistant in a realtime session.

data: bytes

media_type: str

class axio.events.BlockEnd(index: int)

Bases: object

The content block at index is complete.

The point at which accumulated ToolInputDelta fragments are guaranteed to parse. Every provider marks it and axio has had no terminator for it until now.

index: int

class axio.events.Citation(index: int, cited_text: str = '', title: str | None = None, url: str | None = None, source_id: str | None = None, start: int | None = None, end: int | None = None, unit: Literal['char', 'byte', 'page', 'block', 'unknown']='unknown', raw: dict[str, ~typing.Any]=<factory>)

Bases: object

A span of generated text attributed to a source.

What Anthropic’s citation shapes, OpenAI’s annotations and Google’s grounding metadata have in common. raw keeps the provider’s whole object for the fields that differ.

cited_text: str

end: int | None

index: int

raw: dict[str, Any]

source_id: str | None

Whatever identifies the source inside this request: a file id, a document index, a chunk id.

start: int | None

title: str | None

unit: Literal['char', 'byte', 'page', 'block', 'unknown']

What start and end count. Stated because the providers disagree — OpenAI counts characters and Google counts bytes — so offsets from different units must never be compared.

url: str | None

class axio.events.Error(exception: 'BaseException')

Bases: object

exception: BaseException

class axio.events.ImageOutput(index: int, data: bytes, media_type: ImageMediaType)

Bases: object

Model generated an image inline (e.g. Nano Banana / Gemini Image).

data: bytes

index: int

media_type: ImageMediaType

class axio.events.IterationEnd(iteration: 'int', stop_reason: 'StopReason', usage: 'Usage', raw: 'str' = '')

Bases: object

iteration: int

raw: str

The provider’s own word for why it stopped, where it differs from what axio calls it. Kept so a caller can act on a reason this vocabulary has no name for.

stop_reason: StopReason

usage: Usage

class axio.events.IterationStart(iteration: int, id: str | None = None, model: str | None = None)

Bases: object

One provider request has begun.

model is the model that actually served the turn, which need not be the one asked for. Server-side fallback, sticky routing and dated-snapshot resolution all substitute a different model at a different price. A cost lookup therefore keys off this rather than off the request.

id: str | None

iteration: int

model: str | None

class axio.events.ProviderEvent(provider: str, kind: str, data: dict[str, Any], index: int | None = None)

Bases: object

A provider payload axio does not model, forwarded verbatim.

data is the provider’s own JSON object exactly as it was parsed: no renaming, no coercion, no filtering.

How completely a transport forwards depends on what its stream names. A stream that names each event has a reader. That reader forwards every payload it does not interpret, so nothing is dropped. A stream with no discriminator — one shape per payload, read field by field — has no such catch. Those transports forward the parts they know how to name. A field the provider adds inside a payload reaches nobody until someone reads it.

A consumer that does not recognise (provider, kind) ignores it.

data: dict[str, Any]

index: int | None

The content-block or output index, where the payload carries one.

kind: str

The provider’s own discriminator, verbatim, never a name axio invented: matching on it is matching the vocabulary the provider publishes.

provider: str

Which transport produced it: "anthropic", "openai", "google". The Codex transport reads its stream through the shared Responses reader, so its events say "openai" rather than naming a fourth provider.

class axio.events.ProviderOutput(provider: str, kind: str, data: dict[str, Any], index: int, id: str = '')

Bases: object

One output item of this turn that the next request has to send back unread.

ProviderEvent forwards what a caller may want to watch; this one names what the turn is not complete without. An endpoint that keeps no history of its own — store=False on the Responses API — expects every item it produced back on the next request, including those from the tools it ran itself: a web search, a file search, a code interpreter, and item types that do not exist yet. Read as news rather than as content, they were watched and dropped, and the next request was missing what the model had answered from.

The agent stores it as a ProviderBlock, and the transport that speaks the same protocol replays data verbatim.

data: dict[str, Any]

The item exactly as it arrived, never interpreted here.

id: str

index: int

The output index this item occupied. Declared in the same position as on ProviderEvent, whose first three fields these are: with the two orders crossed, one built positionally in the shape of the other assigned every field to the wrong name.

kind: str

The item’s own type, verbatim.

provider: str

Which protocol produced it, in the names ProviderEvent uses.

class axio.events.ReasoningDelta(index: 'int', delta: 'str')

Bases: object

delta: str

index: int

class axio.events.ReasoningSignature(index: int, signature: str, redacted: bool = False, id: str = '', provider: str = '')

Bases: object

Opaque proof that a reasoning block is the provider’s own, to be replayed unaltered.

Anthropic refuses a returned thinking block whose signature is missing or changed, and Google publishes a MISSING_THOUGHT_SIGNATURE finish reason for the same failure. Never inspect, decode, re-encode or truncate data.

id: str

How the provider names the block this proves, where it names them. Replayed beside the proof, because a provider that identifies reasoning by id refuses the pair without it.

index: int

provider: str

Which protocol issued this proof, in the names ProviderEvent uses.

redacted: bool

True where the payload replaces the reasoning text instead of accompanying it.

signature: str

class axio.events.Refusal(index: int, text: str = '', spoken: bool = True, category: str | None = None, blocked_input: bool = False, raw: dict[str, ~typing.Any]=<factory>)

Bases: object

The model declined, or the provider blocked the turn.

This is deliberately not a TextDelta. As ordinary assistant text, or as an empty turn that succeeded, a refusal is indistinguishable from an answer and no consumer can act on it.

blocked_input: bool

True where the provider rejected the prompt rather than the answer, so nothing was generated and sending the same prompt again cannot succeed.

category: str | None

The provider’s own category, verbatim. Not normalised: the taxonomies do not overlap, and a mapping between them would state something no provider says.

index: int

raw: dict[str, Any]

spoken: bool

Whether text is the model’s own words.

True on the endpoints that stream a refusal as output content, which is what OpenAI and the Responses API do. False where the text is the provider explaining why it stopped: Anthropic sends stop_details.explanation, which its own schema documents as unstable and not to be parsed, and the model generated nothing at all.

The agent stores either kind as the turn’s text, because a stored turn with no content is refused by the next request and the explanation is the only account of the decline there is. A consumer that renders the two differently — the model declining, against the provider reporting a block — reads this.

text: str

class axio.events.SessionEndEvent(stop_reason: 'StopReason', total_usage: 'Usage')

Bases: object

stop_reason: StopReason

total_usage: Usage

class axio.events.SpeechStarted

Bases: object

Server VAD detected the user started speaking (realtime).

class axio.events.SpeechStopped

Bases: object

Server VAD detected the user stopped speaking (realtime).

type axio.events.StreamEvent = ReasoningDelta | ReasoningSignature | TextDelta | TextSignature | Refusal | Citation | ImageOutput | AudioOutput | VideoOutput | ToolUseStart | ToolInputDelta | ToolFieldStart | ToolFieldDelta | ToolFieldEnd | ToolOutputDelta | ToolResult | BlockEnd | IterationStart | IterationEnd | Error | ProviderEvent | ProviderOutput | SessionEndEvent | AudioOutputDelta | TranscriptDelta | SpeechStarted | SpeechStopped | TurnComplete

class axio.events.TextDelta(index: 'int', delta: 'str')

Bases: object

delta: str

index: int

class axio.events.TextSignature(index: int, signature: str, provider: str = '')

Bases: object

Opaque proof that a block of answer text is the provider’s own, to be replayed unaltered.

Google signs the part it issued the proof for, and answer text is one such part. The proof belongs to the text block, not to the reasoning or the call beside it: replayed on another part it proves nothing, and the turn fails with MISSING_THOUGHT_SIGNATURE. Never inspect, decode, re-encode or truncate data.

Emitted after the text it signs, never before.

index: int

provider: str

Which protocol issued this proof, in the names ProviderEvent uses.

signature: str

class axio.events.ToolFieldDelta(index: 'int', tool_use_id: 'ToolCallID', key: 'str', text: 'str')

Bases: object

index: int

key: str

text: str

tool_use_id: ToolCallID

class axio.events.ToolFieldEnd(index: 'int', tool_use_id: 'ToolCallID', key: 'str')

Bases: object

index: int

key: str

tool_use_id: ToolCallID

class axio.events.ToolFieldStart(index: 'int', tool_use_id: 'ToolCallID', key: 'str')

Bases: object

index: int

key: str

tool_use_id: ToolCallID

class axio.events.ToolInputDelta(index: 'int', tool_use_id: 'ToolCallID', partial_json: 'str')

Bases: object

index: int

partial_json: str

tool_use_id: ToolCallID

class axio.events.ToolOutputDelta(tool_use_id: 'ToolCallID', name: 'ToolName', key: 'str', delta: 'str')

Bases: object

delta: str

key: str

name: ToolName

tool_use_id: ToolCallID

class axio.events.ToolResult(tool_use_id: 'ToolCallID', name: 'ToolName', is_error: 'bool', content: 'str' = '', input: 'dict[str, Any]'=<factory>)

Bases: object

content: str

input: dict[str, Any]

is_error: bool

name: ToolName

tool_use_id: ToolCallID

class axio.events.ToolUseStart(index: 'int', tool_use_id: 'ToolCallID', name: 'ToolName', signature: 'str' = '', provider: 'str' = '')

Bases: object

index: int

name: ToolName

provider: str

Which protocol issued signature, in the names ProviderEvent uses. Stored with the proof so a session that changes transport does not replay it to one that never made it.

signature: str

Opaque proof that this call is the model’s own, where the provider issues one for the call rather than for the reasoning beside it. Replayed on the call itself: attached to a reasoning block instead, the call comes back unsigned and the provider refuses the turn.

tool_use_id: ToolCallID

class axio.events.TranscriptDelta(role: Literal['user', 'assistant'], delta: str)

Bases: object

Live transcript delta — server-side STT of user mic, or assistant speech transcription, depending on role.

delta: str

role: Literal['user', 'assistant']

class axio.events.TurnComplete(stop_reason: StopReason, usage: Usage | None = None)

Bases: object

Assistant turn finished in a realtime session. stop_reason may be StopReason.tool_use to signal that pending tool calls should run before the next turn starts.

stop_reason: StopReason

usage: Usage | None

class axio.events.VideoOutput(index: int, data: bytes, media_type: VideoMediaType)

Bases: object

Model generated a video inline.

data: bytes

index: int

media_type: VideoMediaType

Protocols

Transport discovery is available to user code through axio.transport. discover_transports() returns a TransportCatalog; use require(name) to get a TransportEntry, open() to enter its lifetime, and select_model() to choose a model. See Choose a transport in application code for runnable application examples, authentication settings, and error handling. These helpers do not require REPL or TUI and are not re-exported from top-level axio.

Transport protocols and application plugin discovery.

class axio.transport.AudioGenTransport(*args, **kwargs)

Bases: Protocol

Generate n non-speech audio samples for a text prompt — music, sound effects, ambient. Returns raw audio bytes (MP3 / WAV / OGG — provider-defined). Distinct from TTSTransport, which is text-to-speech.

async generate_audios(prompt: str, *, model: str | None = None, n: int = 1) → list[bytes]

class axio.transport.CompletionTransport(*args, **kwargs)

Bases: Protocol

stream(messages: list[Message], tools: list[Tool[Any]], system: str) → AsyncIterator[StreamEvent]

class axio.transport.DummyAudioGenTransport(*args, **kwargs)

Bases: DummyTransport, AudioGenTransport

async generate_audios(prompt: str, *, model: str | None = None, n: int = 1) → list[bytes]

class axio.transport.DummyCompletionTransport(*args, **kwargs)

Bases: DummyTransport, CompletionTransport

stream(messages: list[Message], tools: list[Tool[Any]], system: str) → AsyncIterator[StreamEvent]

class axio.transport.DummyEmbeddingTransport(*args, **kwargs)

Bases: DummyTransport, EmbeddingTransport

embed(texts: list[str]) → Any

class axio.transport.DummyImageEditTransport(*args, **kwargs)

Bases: DummyTransport, ImageEditTransport

async edit_images(prompt: str, *, images: list[ImageBlock], mask: ImageBlock | None = None, model: str | None = None, n: int = 1) → list[bytes]

class axio.transport.DummyImageGenTransport(*args, **kwargs)

Bases: DummyTransport, ImageGenTransport

async generate_images(prompt: str, *, model: str | None = None, n: int = 1) → list[bytes]

class axio.transport.DummySTTTransport(*args, **kwargs)

Bases: DummyTransport, STTTransport

async transcribe(audio: bytes, media_type: str = 'audio/wav') → str

class axio.transport.DummyTTSTransport(*args, **kwargs)

Bases: DummyTransport, TTSTransport

synthesize(text: str, *, voice: str | None = None) → AsyncIterator[bytes]

class axio.transport.DummyTransport

Bases: object

Placeholder transport that fails loudly if actually used.

Assign this as the default transport when constructing agent prototypes that will be configured later via agent.copy(transport=real_transport).

Example:

from axio.agent import Agent
from axio.transport import DummyCompletionTransport

researcher = Agent(
    system="You are a research assistant...",
    transport=DummyCompletionTransport(),
)

# At runtime, swap in the real transport:
active = researcher.copy(transport=OpenAITransport())
result = await active.run(task, context)

class axio.transport.DummyVideoGenTransport(*args, **kwargs)

Bases: DummyTransport, VideoGenTransport

async generate_videos(prompt: str, *, model: str | None = None, n: int = 1, image: bytes | None = None, duration_seconds: int | None = None, aspect_ratio: str | None = None) → list[bytes]

class axio.transport.EmbeddingTransport(*args, **kwargs)

Bases: Protocol

async embed(texts: list[str]) → list[list[float]]

class axio.transport.ImageEditTransport(*args, **kwargs)

Bases: Protocol

Edit supplied images according to a prompt, returning complete image bytes.

Sources are required. Implementations must reject unsupported masks instead of ignoring them; supported formats and counts are provider-specific.

async edit_images(prompt: str, *, images: list[ImageBlock], mask: ImageBlock | None = None, model: str | None = None, n: int = 1) → list[bytes]

class axio.transport.ImageGenTransport(*args, **kwargs)

Bases: Protocol

Generate n image samples for a text prompt. Returns raw image bytes (PNG / JPEG / WebP — provider-defined).

async generate_images(prompt: str, *, model: str | None = None, n: int = 1) → list[bytes]

class axio.transport.RealtimeSession(*args, **kwargs)

Bases: Protocol

Active duplex realtime session — bidirectional audio / text / tools.

Returned by RealtimeTransport.connect(). Events from the provider arrive on events(). User input is pushed via send().

async close() → None

Tear down the session and release resources.

async commit() → None

Signal end-of-utterance for manual VAD; no-op with server VAD.

events() → AsyncIterator[StreamEvent]

Async iterator over server events for the lifetime of this session.

async interrupt() → None

Abort in-flight assistant generation (e.g. user interrupted).

async send(content: ContentBlock | list[ContentBlock]) → None

Append user content (audio chunk, text, image) to the input buffer.

async send_tool_result(tool_use_id: ToolCallID, name: ToolName, content: str | list[ContentBlock]) → None

Deliver a tool’s result to the provider so generation can resume.

name is included because some providers (e.g. Gemini Live) require the tool name alongside the call id. OpenAI realtime can ignore it.

class axio.transport.RealtimeTransport(*args, **kwargs)

Bases: Protocol

Provider that supports duplex realtime sessions (e.g. OpenAI Realtime, Gemini Live). Distinct from CompletionTransport because the interaction is bidirectional, not request/response.

async connect(*, system: str, tools: list[Tool[Any]], voice: str | None = None, input_audio_format: str = 'audio/pcm;rate=16000', output_audio_format: str = 'audio/pcm;rate=24000') → RealtimeSession

class axio.transport.STTTransport(*args, **kwargs)

Bases: Protocol

async transcribe(audio: bytes, media_type: str = 'audio/wav') → str

class axio.transport.TTSTransport(*args, **kwargs)

Bases: Protocol

synthesize(text: str, *, voice: str | None = None) → AsyncIterator[bytes]

class axio.transport.TransportAuth(*args, **kwargs)

Bases: Protocol

Account operations owned by a transport package, without frontend imports.

account_id(transport: Any) → str

async describe_accounts(credentials: Path | None, transport: Any) → str

async has_accounts(credentials: Path | None = None) → bool

async logout(transport: Any) → None

async prompt_login_if_needed(credentials: Path | None, account: str | None) → None

async select_account(credentials: Path | None, account: str) → None

async sign_in(credentials: Path | None, account: str | None) → str

class axio.transport.TransportCatalog(entries: dict[str, ~axio.transport.TransportEntry]=<factory>, errors: dict[str, str]=<factory>, sources: dict[str, ~axio.plugins.PluginSource]=<factory>)

Bases: object

Loaded entries and failures, including failures explicitly requested by name.

entries: dict[str, TransportEntry]

errors: dict[str, str]

require(name: str) → TransportEntry

sources: dict[str, PluginSource]

class axio.transport.TransportEntry(name: str, cls: Any, display_name: str, env_vars: tuple[str, ...] = (), supports_temperature: bool = True, supports_max_tokens: bool = True, supports_debug: bool = True, supports_thinking: bool = True, auth: TransportAuth | None = None, auto_select: bool = True)

Bases: object

Adapt a plugin object or a legacy transport class to one application API.

auth: TransportAuth | None

auto_select: bool

bind_tools(transport: Any | None) → None

Let the provider connect its optional tool handlers to a transport.

cls: Any

display_name: str

env_vars: tuple[str, ...]

format_error(error: BaseException) → str

classmethod from_target(name: str, target: Any) → TransportEntry

has_credentials(config: Mapping[str, str] | None = None) → bool

name: str

open(*, session: Any, account: str | None = None, credentials: Path | str | None = None, on_auth_refresh: Callable[[...], Any] | None = None, **settings: Any) → AsyncIterator[Any]

Yield a ready transport; the caller owns the shared HTTP session.

Managed factories own catalog loading and cleanup. Legacy constructors receive session and saved settings; this adapter loads their catalog.

select_model(transport: Any, model: ModelSpec) → Any

supports_debug: bool

supports_max_tokens: bool

supports_temperature: bool

supports_thinking: bool

exception axio.transport.TransportError

Bases: RuntimeError

An installed transport could not be loaded, configured, or opened.

class axio.transport.TransportPlugin(*args, **kwargs)

Bases: Protocol

An entry point may expose a plugin object or a transport class.

Plugin objects implement this contract. Legacy classes use TransportEntry’s defaults and can supply the same metadata as class attributes. Factories yield a ready transport with its model catalog loaded.

auth: TransportAuth | None

display_name: str

env_vars: tuple[str, ...]

has_credentials(config: Mapping[str, str] | None = None) → bool

open(*, session: Any, **settings: Any) → AbstractAsyncContextManager[Any]

supports_debug: bool

supports_max_tokens: bool

supports_temperature: bool

supports_thinking: bool

class axio.transport.VideoGenTransport(*args, **kwargs)

Bases: Protocol

Generate n video samples for a text prompt. Returns raw video bytes (MP4 / WebM — provider-defined). Provider-specific knobs (duration, aspect ratio, seed image, etc.) live as extra kwargs on the implementation.

async generate_videos(prompt: str, *, model: str | None = None, n: int = 1, image: bytes | None = None, duration_seconds: int | None = None, aspect_ratio: str | None = None) → list[bytes]

axio.transport.discover_transports() → TransportCatalog

Load axio.transport entries. The first declaration of each name wins.

A failed first declaration reserves its name too. This prevents an unrelated package from silently replacing the requested provider after a load failure.

axio.transport.select_transport_model(transport: Any, model: ModelSpec) → Any

Clone a transport while preserving its session and provider settings.

Permission system: guards that gate tool execution.

class axio.permission.AllowAllGuard

Bases: PermissionGuard

async check(tool: Tool[Any], **kwargs: Any) → dict[str, Any]

class axio.permission.ConcurrentGuard

Bases: PermissionGuard, ABC

Guard with concurrency control.

Subclass and override check(). __call__ acquires the semaphore then delegates to check(). Set concurrency to control parallelism (default 1 - one check at a time).

concurrency: int = 1

class axio.permission.DenyAllGuard

Bases: PermissionGuard

async check(tool: Tool[Any], **kwargs: Any) → dict[str, Any]

class axio.permission.PermissionGuard

Bases: ABC

Gate for tool calls. Return modified kwargs to allow, raise to deny.

Tool invokes guards via await guard(tool, **kwargs).

Guards receive the Tool object and the raw keyword arguments before execution. Return the (possibly modified) dict to allow; raise GuardError to deny. Because guards see all inputs up front, they are also the right place for logging and auditing:

class AuditGuard(PermissionGuard):
    async def check(self, tool: Tool, **kwargs: Any) -> dict[str, Any]:
        logger.info("tool=%s args=%s", tool.name, kwargs)
        return kwargs  # always allow

See examples/agent_swarm/agent_swarm/__main__.py (RoleGuard) for a production example.

abstractmethod async check(tool: Tool[Any], **kwargs: Any) → dict[str, Any]

class axio.selector.ToolSelector(*args, **kwargs)

Bases: Protocol

async select(messages: Iterable[Message], tools: Iterable[Tool[Any]]) → Iterable[Tool[Any]]

Realtime duplex agent: drives a RealtimeSession, dispatches tools concurrently with audio output, exposes a single event stream to the caller.

class axio.realtime.RealtimeAgent(system: str, transport: RealtimeTransport, tools: list[Tool[Any]] = <factory>, voice: str | None = None, input_audio_format: str = 'audio/pcm;rate=16000', output_audio_format: str = 'audio/pcm;rate=24000', raise_on_error: bool = True)

Bases: object

Drives a duplex realtime session against a RealtimeTransport.

The agent intercepts ToolUseStart/ToolInputDelta/TurnComplete events to assemble tool calls and dispatches them as background tasks so that streaming audio output from the provider is not blocked. Each tool result is sent back to the session via RealtimeSession.send_tool_result() as soon as the handler returns. interrupt() cancels in-flight tasks.

Lifecycle:

async with RealtimeAgent(system="...", transport=t, tools=[...]) as agent:
    await agent.send(AudioBlock(data=mic_chunk, media_type="audio/pcm"))
    async for event in agent.events():
        match event:
            case AudioOutputDelta(data=pcm):
                speaker.feed(pcm)
            ...

async close() → None

async commit() → None

async connect() → None

async events() → AsyncIterator[StreamEvent]

Yield events from the session, dispatching tool calls as side-effects.

Tool dispatch runs as background tasks — slow handlers do not block the stream of AudioOutputDelta from the provider.

input_audio_format: str

async interrupt() → None

output_audio_format: str

raise_on_error: bool

Re-raise the exception wrapped in any Error event yielded by the session. Set to False to receive Error events as data and decide what to do per error (transient retry, log-and-continue, etc.).

async send(content: ContentBlock | list[ContentBlock]) → None

property session: RealtimeSession

system: str

tools: list[Tool[Any]]

transport: RealtimeTransport

voice: str | None

Data types

Message: the fundamental unit of conversation history.

class axio.messages.Message(role: "Literal['user', 'assistant', 'system']", content: 'list[ContentBlock]' = <factory>)

Bases: object

content: list[ContentBlock]

classmethod from_dict(data: dict[str, Any]) → Message

role: Literal['user', 'assistant', 'system']

to_dict() → dict[str, Any]

Content blocks: TextBlock, ImageBlock, AudioBlock, ToolUseBlock, ToolResultBlock.

class axio.blocks.AudioBlock(media_type: 'AudioMediaType', data: 'bytes')

Bases: ContentBlock

data: bytes

media_type: AudioMediaType

type axio.blocks.AudioMediaType = Literal['audio/x-aac', 'audio/flac', 'audio/mp3', 'audio/m4a', 'audio/mpeg', 'audio/mpga', 'audio/mp4', 'audio/ogg', 'audio/pcm', 'audio/wav', 'audio/webm']

class axio.blocks.ContentBlock

Bases: object

Base class for all content blocks.

class axio.blocks.ImageBlock(media_type: 'ImageMediaType', data: 'bytes')

Bases: ContentBlock

data: bytes

media_type: ImageMediaType

type axio.blocks.ImageMediaType = Literal['image/jpeg', 'image/png', 'image/gif', 'image/webp']

class axio.blocks.ProviderBlock(provider: str, kind: str, data: dict[str, Any], id: str = '')

Bases: ContentBlock

One item of the turn that only the protocol which produced it can read.

An endpoint that runs its own tools answers with items this vocabulary has no shape for: a web search it ran, a file it read, code it executed, and whatever it adds next. Where the application keeps the history rather than the provider — store=False on the Responses API — the provider expects every one of those items back, in order and exactly once. Dropped, the next request is missing what the model was answering from, and reasoning continuity, tool-call association and anything built on a newer item type break with nothing said.

data is the item exactly as it arrived and is never interpreted here. provider names the protocol that issued it, so it is replayed only to an endpoint that speaks the same one, and kind is the item’s own type, which is all a caller needs to render or count it.

data: dict[str, Any]

id: str

How the provider names this item, where it names them.

kind: str

provider: str

class axio.blocks.ReasoningBlock(text: str = '', signature: str = '', redacted: bool = False, id: str = '', provider: str = '')

Bases: ContentBlock

The model’s own reasoning, kept so the turn can be sent back unaltered.

signature is the provider’s proof that the block is its own. Anthropic refuses a returned thinking block whose signature is missing or changed, and Google reports MISSING_THOUGHT_SIGNATURE for the same failure. A stored turn that dropped the signature cannot be replayed. Never inspect, decode or truncate it.

redacted marks a block whose reasoning the provider withheld. The signature still has to travel, and text is empty.

id: str

How the provider names this block, where it names them. Required beside the signature.

provider: str

Which protocol issued signature: "anthropic", "google", "openai", the same names ProviderEvent uses. The value means nothing outside the protocol that made it — Anthropic sends it as a thinking signature, Google as thoughtSignature, Responses as encrypted_content — so a session that changes transport must not replay it to the next one. Empty says nobody recorded it, which is what a turn stored before this field existed looks like; proof() lets that through rather than dropping proofs it cannot judge.

redacted: bool

signature: str

Out of repr on purpose: this is the one field documented as never to be inspected, and a debug log of the block printed it in full beside everything else.

text: str

class axio.blocks.TextBlock(text: 'str', signature: 'str' = '', provider: 'str' = '')

Bases: ContentBlock

provider: str

Which protocol issued signature. See ReasoningBlock.provider.

signature: str

Opaque proof for this text, where the provider signs a text part. Replayed with it. Kept out of repr with the rest of them: see provider below.

text: str

class axio.blocks.ToolResultBlock(tool_use_id: 'ToolCallID', content: 'str | list[TextBlock | ImageBlock | AudioBlock | VideoBlock]', is_error: 'bool' = False)

Bases: ContentBlock

content: str | list[TextBlock | ImageBlock | AudioBlock | VideoBlock]

is_error: bool

tool_use_id: ToolCallID

class axio.blocks.ToolUseBlock(id: 'ToolCallID', name: 'ToolName', input: 'dict[str, Any]', signature: 'str' = '', provider: 'str' = '')

Bases: ContentBlock

id: ToolCallID

input: dict[str, Any]

name: ToolName

provider: str

Which protocol issued signature. See ReasoningBlock.provider.

signature: str

Opaque proof for this call. Stored rather than held in the transport, which a restart empties.

class axio.blocks.VideoBlock(media_type: 'VideoMediaType', data: 'bytes')

Bases: ContentBlock

data: bytes

media_type: VideoMediaType

type axio.blocks.VideoMediaType = Literal['video/mp4', 'video/mpeg', 'video/mov', 'video/avi', 'video/x-flv', 'video/mpg', 'video/webm', 'video/wmv', 'video/3gpp']

axio.blocks.from_dict(data: dict[str, Any]) → ContentBlock

Deserialize a plain dict to a ContentBlock.

axio.blocks.proof(block: TextBlock | ReasoningBlock | ToolUseBlock, provider: str) → str

This block’s opaque proof, where the protocol about to receive it is the one that issued it.

Each converter reads the same signature field by its own rules, so a session that changed transport sent a proof one provider made to another that never made it. At best the request is refused; at worst the value is accepted somewhere it was never meant to sit, or printed by whatever logs the request.

A block with no recorded provider was stored before anyone recorded one, and the transport reading it is almost always the one that wrote it. Those are replayed, because dropping them loses proofs that are valid and breaks the sessions that already exist.

axio.blocks.replayable(block: ProviderBlock, provider: str) → bool

Whether this opaque item goes back to the endpoint about to be asked.

Unlike a proof, an item with no provider recorded is not replayed: every one of them was made by a reader that names itself, so an empty name is a block from somewhere else entirely.

axio.blocks.to_dict(block: ContentBlock) → dict[str, Any]

axio.blocks.to_dict(block: TextBlock) → dict[str, Any]

axio.blocks.to_dict(block: ImageBlock) → dict[str, Any]

axio.blocks.to_dict(block: AudioBlock) → dict[str, Any]

axio.blocks.to_dict(block: VideoBlock) → dict[str, Any]

axio.blocks.to_dict(block: ReasoningBlock) → dict[str, Any]

axio.blocks.to_dict(block: ToolUseBlock) → dict[str, Any]

axio.blocks.to_dict(block: ProviderBlock) → dict[str, Any]

axio.blocks.to_dict(block: ToolResultBlock) → dict[str, Any]

Serialize a ContentBlock to a plain dict.

Primitive types: ToolName, ToolCallID, StopReason, Usage.

axio.types.INCOMPLETE: Final = frozenset({StopReason.cancelled, StopReason.context_window_exceeded, StopReason.max_tokens, StopReason.unknown})

Reasons that end a run holding an answer that is not finished, and say so nowhere else. end_turn finished one. refusal and error announce themselves through events of their own, and repetition writes its own note into the text it cut. These four say nothing unless the caller shows them, and a truncated answer then reads exactly like a whole one.

class axio.types.StopReason(*values)

Bases: StrEnum

Why the provider stopped generating.

Anything that is not tool_use or pause_turn ends the run. See the match in Agent._run_loop, whose wildcard keeps a member added here from falling through into another paid iteration.

cancelled = 'cancelled'

The caller or the provider stopped the turn before it finished.

context_window_exceeded = 'context_window_exceeded'

The conversation outgrew the model’s window. Truncated, like max_tokens.

end_turn = 'end_turn'

error = 'error'

max_tokens = 'max_tokens'

pause_turn = 'pause_turn'

A server-side tool loop reached its iteration limit. Resumable: the provider expects the assistant content back so it can finish. This is the one reason that does not end the run.

refusal = 'refusal'

The model declined, or the provider blocked the turn. Not an error: the same prompt sent again will be declined again.

repetition = 'repetition'

Axio stopped the turn, because the model was repeating itself. The only reason here the provider did not give. Reported as end_turn, a caller could not tell an answer the model finished from one cut off mid-word, which is the same objection this vocabulary raises against reading a truncated response as a whole one.

tool_use = 'tool_use'

unknown = 'unknown'

The provider said something this vocabulary does not have. Terminal, and it vouches for nothing. Named rather than folded into one of the others, because each of those claims something the provider did not say: that the turn finished, that it was truncated, or that the transport broke. IterationEnd.raw carries the word itself.

type axio.types.ToolCallID = str

type axio.types.ToolName = str

class axio.types.Usage(input_tokens: int, output_tokens: int, *, cache_read_tokens: int = 0, cache_write_tokens: int = 0, reasoning_tokens: int = 0)

Bases: object

Token counts for one provider request.

The rule: input_tokens and output_tokens are always inclusive grand totals, and every other field is a disjoint slice of one of them:

cache_read_tokens + cache_write_tokens  <=  input_tokens
reasoning_tokens                        <=  output_tokens

Providers disagree about whether their own headline number already contains the slices, and they disagree in opposite directions. Anthropic counts only the tokens after the last cache breakpoint, so its cache counts have to be added. Google reports thinking beside the candidates rather than inside them. Each transport adds or does not add to satisfy the rule here, so nothing downstream has to know which provider answered.

Counts only, never money. A cached token and a written one bill at different multipliers, so a caller that wants cost multiplies these slices by its own per-model rates. A zero slice means the provider billed none of it, or reported no breakdown at all. Axio cannot tell those apart.

property answer_tokens: int

Output that was the answer rather than reasoning.

cache_read_tokens: int

The slice of input_tokens served from cache, billed at a discount.

cache_write_tokens: int

The slice of input_tokens written to cache, billed at a premium. Disjoint from the read.

input_tokens: int

output_tokens: int

reasoning_tokens: int

The slice of output_tokens spent on reasoning the caller never sees.

classmethod reported(input_tokens: int, output_tokens: int, *, cache_read_tokens: int = 0, cache_write_tokens: int = 0, reasoning_tokens: int = 0) → Usage

What a provider said, held to the rule above rather than trusted to follow it.

Constructed directly, a provider that reports a slice larger than the total it belongs to fails __post_init__, and a whole answer is lost over an accounting discrepancy. Every transport reads its provider’s numbers through here instead.

A slice that outgrew its total means the total was reported without it. The total is therefore raised to hold the slices, which keeps the tokens the provider billed for; the other repair, cutting the slice down, throws them away and under-reports the cost.

property total_tokens: int

property uncached_input_tokens: int

Input the provider had to read in full, which is what most of the bill is.

axio.types.stop_reason_from(raw: str, table: Mapping[str, StopReason], *, provider: str) → StopReason

What a provider’s own stop value means here, or unknown where the table does not say.

There is no fifth answer to give. Folded into end_turn it claims the turn finished, into max_tokens that it was truncated, into error that the transport broke; raising throws away an answer the caller has already read. unknown says only what is true, and IterationEnd.raw carries the provider’s own word for the caller to act on.

Transport-agnostic model types: Capability, ModelSpec, ModelRegistry.

class axio.models.Capability(*values)

Bases: StrEnum

audio = 'audio'

embedding = 'embedding'

image_editing = 'image_editing'

image_generation = 'image_generation'

json_mode = 'json_mode'

reasoning = 'reasoning'

structured_outputs = 'structured_outputs'

text = 'text'

tool_use = 'tool_use'

video = 'video'

video_generation = 'video_generation'

vision = 'vision'

class axio.models.ModelRegistry(models: Iterable[ModelSpec] | None = None)

Bases: MutableMapping[str, ModelSpec]

by_capability(*caps: Capability) → ModelRegistry

by_cost(*, output: bool = False, desc: bool = False) → ModelRegistry

Return registry ordered by cost (input by default, output if output=True).

by_prefix(prefix: str) → ModelRegistry

clear() → None.  Remove all items from D.

first() → ModelSpec

ids() → list[str]

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

last() → ModelSpec

search(*q: str) → ModelRegistry

search by parts of id

values() → an object providing a view on D's values

class axio.models.ModelSpec(id: 'str', capabilities: 'frozenset[Capability]' = frozenset(), max_output_tokens: 'int' = 8192, context_window: 'int' = 128000, input_cost: 'float' = 0.0, output_cost: 'float' = 0.0)

Bases: object

capabilities: frozenset[Capability]

context_window: int

id: str

input_cost: float

max_output_tokens: int

output_cost: float

Field metadata for tool handler functions - lightweight replacement for pydantic.Field.

axio.field.Field(description: str = '', default: Any = MISSING, ge: int | float | None = None, le: int | float | None = None) → FieldInfo

Annotate a handler parameter with metadata (description, default, constraints).

Usage:

async def search(
    query: Annotated[str, Field(description="Search query")],
    limit: Annotated[int, Field(default=10, ge=1, le=100)],
) -> str: ...

class axio.field.FieldInfo(description: str = '', default: Any = MISSING, ge: int | float | None = None, le: int | float | None = None, strict: bool = False)

Bases: object

Metadata attached to a handler parameter via Annotated[T, FieldInfo(...)].

default: Any = MISSING

description: str = ''

ge: int | float | None = None

le: int | float | None = None

strict: bool = False

validate(value: Any, name: str, hint: Any) → None

Validate value against this FieldInfo’s constraints, raising if invalid.

axio.field.MISSING: Final[MissingSentinel] = MISSING

Sentinel value meaning “this field has no default and is required”.

class axio.field.MissingSentinel

Bases: object

Singleton sentinel - distinguishes ‘no default’ from None.

axio.field.StrictStr

Drop-in replacement for from pydantic import StrictStr. Rejects non-string values (e.g. integers) without coercion.

alias of Annotated[str, FieldInfo(description=, default=MISSING, ge=None, le=None, strict=True)]

axio.field.bare_type(hint: Any) → type

Return the base Python type, stripping Annotated, Optional, and generic wrappers.

axio.field.check_list_items(value: list[Any], name: str, inner: Any) → None

Raise TypeError when any list element violates the generic item type.

axio.field.check_scalar(value: Any, name: str, b: type, strict: bool) → None

Raise TypeError when value does not satisfy the scalar type b.

axio.field.check_type(value: Any, name: str, inner: Any, *, strict: bool) → None

Dispatch type validation for value against inner (already unwrapped).

axio.field.get_field_info(annotation: Any) → FieldInfo | None

Extract a FieldInfo from an Annotated annotation, or return None.

axio.field.is_classvar(annotation: Any) → bool

Return True if annotation is ClassVar or ClassVar[X].

axio.field.unwrap_hint(hint: Any) → tuple[Any, bool]

Strip Annotated and Optional wrappers.

Returns (inner_type, is_optional) where inner_type has no Annotated or Union-with-None wrappers and is_optional is True when None was one of the Union members.

JSON-schema builder for plain async handler functions.

Produces clean schemas with no "title" keys - no post-processing needed in transports.

axio.schema.build_tool_schema(fn: Any, hints: dict[str, Any] | None = None) → dict[str, Any]

Return a JSON schema object for fn (a callable or class).

The schema has the form:

{
    "type": "object",
    "properties": {"field": {"type": "string"}, ...},
    "required": ["field", ...]   # only when non-empty
}

No "title" keys are emitted anywhere in the schema.

Parameters

fn:

A plain async function or a class whose annotations define the fields.

hints:

Pre-computed get_type_hints(fn, include_extras=True) result. When supplied, the call to get_type_hints is skipped.

axio.schema.property_schema(annotation: Any) → dict[str, Any]

Recursively convert a Python type annotation to a JSON schema fragment.

axio.schema.strip_title(schema: dict[str, Any]) → dict[str, Any]

The schema without its title keywords, at every depth.

build_tool_schema writes none, so this is for a schema that came from somewhere else: a pydantic model names every model and field, no provider reads those names, and a large tool set pays for them on every request.

Only the keywords that hold schemas are walked. const, default, enum and examples hold values the caller declared, so a value with a title field of its own survives: walked as schemas, an enum of objects came out as a list of empty ones.

Credentials

axio.credentials provides typed credential storage without provider dependencies. Credentials[T] supplies locking, serialization, and record selection for the file and SQLite backends. CredentialStore[T] is the protocol for other backends.

A provider supplies its dataclass and a default filename. Mark exactly one string field with metadata={"credential_key": True}. The key can be inside a nested dataclass. Use repr=False for secret fields.

For an additive schema change, a new field with a default or default factory can declare metadata={"credential_optional": True}. Older records that omit it then use that default. All unmarked fields remain required even if their dataclass has a default; missing recovery markers must not silently reset operation state. Unknown fields and incorrectly typed values are still rejected.

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from pathlib import Path
from tempfile import TemporaryDirectory

from axio.credentials import FileCredentialStore


@dataclass(frozen=True, slots=True)
class ServiceAccount:
    name: str = field(metadata={"credential_key": True})
    api_key: str = field(repr=False)


class ServiceCredentials(FileCredentialStore[ServiceAccount]):
    filename = "service.auth"


async def example(path: Path) -> None:
    store = ServiceCredentials(path)
    async with store.lock("work"):
        await store.save(ServiceAccount("work", "example-key"))
    async with store.lock("work"):
        account = await store.load("work")  # ServiceAccount | None
        assert account == ServiceAccount("work", "example-key")


with TemporaryDirectory() as directory:
    asyncio.run(example(Path(directory) / "service.auth"))

For a caller-selected path, FileCredentialStore[ServiceAccount](path) also retains the record type. Alternatively, pass record_type=ServiceAccount. SQLiteCredentialStore[ServiceAccount](database_path) has the same record API. The default file path is $XDG_CONFIG_HOME/axio/<filename>, falling back to ~/.config/axio/<filename> when XDG_CONFIG_HOME is unset or relative.

The shared codec uses asdict() and restores nested dataclasses, optional values, unions, literals, enums, lists, tuples, sets, frozensets, and dictionaries with string keys. Scalar types are checked without coercing strings or booleans into numbers. Non-finite numbers, missing fields, and unknown fields are rejected. Dataclass __post_init__ methods can check relationships between fields; raise ValueError for invalid data. Codec errors do not include stored values.

Call load() and save() under the corresponding lock(). Each save commits independently, even if a later operation fails. Cancellation waits for an active write before releasing the lock. Keep locks held across credential renewal and its writes. Local backends require POSIX permissions and flock; stored data is not encrypted. Preserve the adjacent .locks directory while processes use it.

list_records() returns complete records, including secrets. Applications must remove secret fields before displaying account metadata. select_account() and selected_account() store a preference for future sessions; host_id() retains a stable local identity. The storage layer does not perform OAuth or refresh.

For Redis, PostgreSQL, or another backend, implement CredentialStore[T] with lock(key), load(key), and save(record). Use record_id(record) for the key and CredentialCodec(T) for shared validation and serialization. Distributed locks must remain exclusive throughout each operation. Saves must be atomic and durable before returning.

Local backend extensions override read_value() and write_value() with dictionary values. These methods run in worker threads; JSON encoding belongs at the persistence boundary. SQLite opens and closes a connection within each worker-thread operation. Existing ChatGPT JSON records and SQLite databases remain readable, with their original lock names and recovery markers.

class axio.credentials.Credentials(directory: Path, locks: Path, *, record_type: type[T] | None = None)

Shared process locking and cancellation handling for local backends.

Override read_value/write_value to change persistence. These synchronous hooks run in a worker thread. Each write must commit before returning. load/save require lock(key_id) in the same asyncio task. Locks are not reentrant. The local filesystem must support flock and durable writes.

property codec: CredentialCodec[T]

Resolve T from a typed instance, a typed subclass, or record_type.

async host_id() → str

Create or load a stable host ID under an independent process lock.

async list_records() → list[T]

Read a typed snapshot. Records can contain secrets; do not print them.

abstractmethod read_value(key: str) → dict[str, Any] | None

Read the latest committed value. Called under the corresponding lock.

read_values() → dict[str, Any]

Read a snapshot for account discovery. Optional for custom adapters.

async select_account(key_id: str) → None

Select an existing registration for future sessions.

async selected_account() → str | None

Read the selected record key. This does not change active sessions.

abstractmethod write_value(key: str, value: dict[str, Any]) → None

Commit a whole value durably. Called under the corresponding lock.

class axio.credentials.CredentialStore(*args, **kwargs)

Storage contract implemented by the application, with any backend.

lock() excludes all writers for one record key, including other processes and hosts that share these credentials. Keep it held across HTTP requests. A distributed lease must remain exclusive until exit; a lease that silently expires can cause refresh-token reuse.

load() reads the latest record, never a stale process-local cache. save() atomically and durably replaces the whole record BEFORE returning. Each save is committed separately, even inside lock(); lock() is not a database transaction that rolls back writes on exit. Recovery markers must survive a process crash during the subsequent HTTP exchange.

Call load/save under lock. Report failures as CredentialError with safe messages. Protect credentials through filesystem permissions, database access controls, or equivalent. Keep secrets out of logs and repr. Retain the host ID separately across restarts and sign-outs. Axio selects neither a backend nor a storage path.

class axio.credentials.FileCredentialStore(path: str | Path | None = None, *, record_type: type[T] | None = None)

Store registrations in one JSON file with atomic replacement.

Defaults to $XDG_CONFIG_HOME/axio/<filename> or ~/.config/axio/<filename>. Creates directories with mode 0700 and files with mode 0600. Rejects files with unsafe permissions, symlinks, and hard links. Keep the adjacent .locks directory intact while any process uses the store. Credentials are protected by filesystem access controls, not encrypted at rest.

read_value(key: str) → dict[str, Any] | None

Read the latest committed value. Called under the corresponding lock.

read_values() → dict[str, Any]

Read a snapshot for account discovery. Optional for custom adapters.

write_value(key: str, value: dict[str, Any]) → None

Commit a whole value durably. Called under the corresponding lock.

class axio.credentials.SQLiteCredentialStore(path: str | Path, *, record_type: type[T] | None = None)

Store registrations in a dedicated SQLite database in a private directory.

Each load opens a mode=ro connection, reads, and closes in one worker-thread call. Each write opens its own connection, commits, and closes. The adjacent <database>.locks directory coordinates refresh independently of SQL commits. Connections never span an HTTP request. No close() call is necessary.

connect(*, readonly: bool) → Connection

Open one connection. The calling worker thread must close it.

read_value(key: str) → dict[str, Any] | None

Read the latest committed value. Called under the corresponding lock.

read_values() → dict[str, Any]

Read a snapshot for account discovery. Optional for custom adapters.

table_name(conn: Connection) → str

Keep using an existing legacy table so old and new processes share writes.

write_value(key: str, value: dict[str, Any]) → None

Commit a whole value durably. Called under the corresponding lock.

class axio.credentials.CredentialCodec(record_type: type[T])

Versioned dataclass records with safe errors and strict type checks.

Utilities

AgentStream: async iterator wrapper over the agent event generator.

class axio.stream.AgentStream(generator: AsyncGenerator[StreamEvent, None])

Bases: object

async aclose() → None

async get_final_text() → str

Everything the turn said, and nothing about whether it finished saying it.

A run ending on one of INCOMPLETE returns a truncated answer that reads exactly like a whole one, because a str has nowhere to put the reason. It is logged as a warning here, and get_session_end() carries it for a caller that needs to branch on it. Error still raises, so a broken turn is never returned as an answer.

async get_session_end() → SessionEndEvent

Load Agent prototypes from declarative config files (TOML, JSON, INI).

Each file describes one agent. The loader scans a directory, parses every recognised file, resolves tool names against a caller-supplied toolbox, and returns a dict[str, tuple[str, Agent]] that matches the shape used by agent_swarm and similar orchestrators.

Supported formats

TOML:

name = "architect"
description = "System design and interface specs"
max_iterations = 100
tools = ["read_file", "write_file"]

[system]
text = """
You are an expert software architect...
"""

JSON:

{
  "name": "architect",
  "description": "System design and interface specs",
  "max_iterations": 100,
  "tools": ["read_file", "write_file"],
  "system": "You are an expert software architect..."
}

INI:

[agent]
name = architect
description = System design and interface specs
max_iterations = 100
tools = read_file, write_file

[system]
text = You are an expert software architect...

In all formats name falls back to the file stem when omitted. system may be a plain string or a {"text": "..."} mapping (TOML/JSON). INI tools are comma-separated.

Custom sources

Subclass AgentLoader and implement load() - the base load_file() will handle reading the file and calling your implementation automatically:

class DbAgentLoader(AgentLoader):
    def load(self, content: str) -> AgentSpec:
        row = json.loads(content)
        return AgentSpec(name=row["name"], ...)

class axio.agent_loader.AgentContext

Bases: TypedDict

agent_name: str

context_factory: Callable[[], ContextStore]

on_event: Callable[[str, StreamEvent], None] | None

proto: Agent

transport: CompletionTransport

class axio.agent_loader.AgentLoader

Bases: object

Base class for format-specific agent loaders.

Subclasses implement load() to parse a raw string. The source of that string is entirely up to the caller - files, databases, HTTP, etc. load_file() is provided on the base class and calls load() automatically.

extensions: tuple[str, ...] = ()

load(content: str) → AgentSpec

Parse content and return an AgentSpec.

name defaults to "" when not present. load_file() patches it from the file stem after calling this method.

load_file(path: Path) → AgentSpec

Read path and delegate to load(), patching name from the stem.

Any ValueError raised by load() is re-raised with path prepended so errors are easy to locate.

scan(directory: Path, toolbox: Mapping[str, Tool[Any]] = mappingproxy({})) → dict[str, tuple[str, Agent]]

Scan directory for agent files and return a name→(description, agent) dict.

class axio.agent_loader.AgentSpec(name: str, description: str, system: str, max_iterations: int = 50, tools: tuple[str, ...] = (), model: str | None = None)

Bases: object

Parsed agent definition before transport/tools are injected.

description: str

max_iterations: int = 50

model: str | None = None

name: str

system: str

to_agent(toolbox: Mapping[str, Tool[Any]] = mappingproxy({})) → Agent

Return a prototype Agent with toolbox tools attached.

The agent uses DummyCompletionTransport - call agent.copy(transport=real_transport) before running it.

Raises KeyError if any name in self.tools is absent from toolbox.

tools: tuple[str, ...] = ()

class axio.agent_loader.IniAgentLoader

Bases: AgentLoader

Load agent definitions from .ini files.

Expected sections: [agent] for metadata, [system] for the prompt. The tools key is comma-separated. Multiline system prompts work via standard ConfigParser continuation (indent subsequent lines).

extensions: tuple[str, ...] = ('ini',)

load(content: str) → AgentSpec

Parse content and return an AgentSpec.

name defaults to "" when not present. load_file() patches it from the file stem after calling this method.

class axio.agent_loader.JsonAgentLoader

Bases: AgentLoader

Load agent definitions from .json files.

extensions: tuple[str, ...] = ('json',)

load(content: str) → AgentSpec

Parse content and return an AgentSpec.

name defaults to "" when not present. load_file() patches it from the file stem after calling this method.

class axio.agent_loader.MultiFormatLoader(loaders: list[AgentLoader] = <factory>)

Bases: object

Aggregate loader that handles TOML, JSON, and INI files.

loaders: list[AgentLoader]

scan(directory: Path, toolbox: Mapping[str, Tool[Any]] = mappingproxy({})) → dict[str, tuple[str, Agent]]

Scan directory with all registered loaders, later loaders win on name collision.

class axio.agent_loader.TomlAgentLoader

Bases: AgentLoader

Load agent definitions from .toml files.

extensions: tuple[str, ...] = ('toml',)

load(content: str) → AgentSpec

Parse content and return an AgentSpec.

name defaults to "" when not present. load_file() patches it from the file stem after calling this method.

async axio.agent_loader.agent_tool(task: ~typing.Annotated[str, ~axio.field.FieldInfo(description=Full task instructions., default=MISSING, ge=None, le=None, strict=False)]) → str

Delegate a task to a sub-agent.

axio.agent_loader.load_agents(directory: Path, toolbox: Mapping[str, Tool[Any]] = mappingproxy({})) → dict[str, tuple[str, Agent]]

Scan directory for .toml, .json, and .ini agent files.

Returns dict[name, (description, agent)] - same shape as the AGENTS registry used in agent_swarm and similar examples.

Example:

from pathlib import Path
from axio.agent_loader import load_agents

AGENTS = load_agents(
    Path(__file__).parent / "roles",
    toolbox={"read_file": read_file_tool, "write_file": write_file_tool},
)

axio.agent_loader.load_agents_from_dir(directory: ~pathlib.Path, transport: ~axio.transport.CompletionTransport, context_factory: ~collections.abc.Callable[[], ~axio.context.ContextStore] = <class 'axio.context.MemoryContextStore'>, on_event: ~collections.abc.Callable[[str, ~axio.events.StreamEvent], None] | None = None) → list[Tool[AgentContext]]

axio.agent_loader.make_agent_tools(agents: dict[str, tuple[str, ~axio.agent.Agent]], transport: ~axio.transport.CompletionTransport, context_factory: ~collections.abc.Callable[[], ~axio.context.ContextStore] = <class 'axio.context.MemoryContextStore'>, on_event: ~collections.abc.Callable[[str, ~axio.events.StreamEvent], None] | None = None, agent_name_prefix: str = '') → list[Tool[AgentContext]]

Convert each agent into its own Tool.

Each tool is named after the agent and accepts a single task field. Runtime dependencies (transport, context factory, event callback) are stored in context and injected on each call.

Parameters

agents:

dict[name, (description, prototype_agent)] - e.g. the value returned by load_agents().

transport:

Transport assigned to the selected agent via agent.copy().

context_factory:

Called once per invocation to produce a fresh ContextStore. Defaults to MemoryContextStore.

on_event:

Optional callback receiving (agent_name, event) for every StreamEvent the agent emits.

agent_name_prefix:

Prefix to prepend to each agent name.

Example:

from axio.agent_loader import load_agents, make_agent_tools

agents = load_agents(Path("roles"), toolbox={"read_file": read_file_tool})
tools = make_agent_tools(agents, transport=my_transport)
orchestrator = Agent(system="...", transport=my_transport, tools=tools)

Incremental streaming parser for tool call JSON arguments.

Feeds partial JSON chunks (from ToolInputDelta.partial_json) and emits structured ToolField* events as top-level object fields are discovered.

Top-level string values are decoded (escape sequences resolved, quotes stripped). All other top-level values are emitted as raw JSON fragments.

class axio.tool_args.State(*values)

Bases: IntEnum

AFTER = 7

COLON = 3

ESC = 8

INIT = 0

KEY = 2

OBJ = 1

RAW = 6

STR = 5

UESC = 9

VAL = 4

class axio.tool_args.ToolArgStream(tool_use_id: str, index: int = 0)

Bases: object

O(1)-per-character streaming parser for tool argument JSON.

Usage:

stream = ToolArgStream("call_1")
events = stream.feed('{"path":"/tmp/f')
# [ToolFieldStart(0, "call_1", "path"), ToolFieldDelta(0, "call_1", "path", "/tmp/f")]
events = stream.feed('oo.py"}')
# [ToolFieldDelta(0, "call_1", "path", "oo.py"), ToolFieldEnd(0, "call_1", "path")]

property current_key: str

The field currently being streamed, or "".

property done: bool

Whether the top-level JSON object has been fully parsed.

feed(chunk: str) → list[ToolFieldEvent]

Process a partial JSON chunk and return any field events produced.

type axio.tool_args.ToolFieldEvent = ToolFieldStart | ToolFieldDelta | ToolFieldEnd

Context compaction: summarise old history to stay within token limits.

class axio.compaction.AutoCompactStore(store: ContextStore, transport: CompletionTransport, *, keep_recent: int = 6, max_tokens: int | None = None, threshold: float = 0.75)

Bases: ContextStore

Delegating ContextStore wrapper that auto-compacts the inner store when token usage exceeds a threshold. Works with any ContextStore backend - MemoryContextStore, SQLiteContextStore, etc.

The threshold defaults to 75 % of transport.model.context_window (read via getattr; falls back to 128 000 if the transport has no model attribute). Pass max_tokens explicitly to override.

Compaction fires from add_context_tokens(), which the agent loop calls immediately after IterationEnd. input_tokens there equals the real context size sent to the model in that iteration.

Internally, _do_compact() forks the inner store before calling compact_context. The fork acts as a stable snapshot. The summarisation agent reads from it while the live store remains writable. The live store is only cleared and repopulated after the (async) summary call returns.

Example:

from axio.compaction import AutoCompactStore
from axio.context import MemoryContextStore

store = AutoCompactStore(MemoryContextStore(), transport, keep_recent=6)
result = await agent.run(task, store)

async add_context_tokens(input_tokens: int, output_tokens: int) → None

async append(message: Message) → None

async clear() → None

async close() → None

No-op by default.

async fork() → AutoCompactStore

Return an AutoCompactStore wrapping a fork of the inner store.

async get_context_tokens() → tuple[int, int]

Returns (0, 0) by default.

async get_history() → list[Message]

async list_sessions() → list[SessionInfo]

List available sessions. Default: returns a single entry for the current session.

property session_id: str

Lazy-init UUID hex; works without calling super().__init__().

async set_context_tokens(input_tokens: int, output_tokens: int) → None

No-op by default; tokens are silently dropped.

async axio.compaction.compact_context(context: ContextStore, transport: CompletionTransport, *, keep_recent: int = 6, system_prompt: str = 'You are a conversation summarizer. You will see a conversation between a user and an AI assistant, including tool calls and their results. Produce a concise summary preserving: user goals, decisions made, key facts, tool outcomes, and state changes. Write as narrative prose, not as a transcript.') → list[Message] | None

Summarize old messages from context, keeping recent ones verbatim.

Returns a compacted message list ready to populate a fresh store, or None if the history is too short to compact (split <= 0).

The caller is responsible for deciding when to compact (e.g. by tracking usage.input_tokens from IterationEnd events).

Exception hierarchy for axio.

exception axio.exceptions.AxioError

Bases: Exception

Base exception for all axio errors.

exception axio.exceptions.GuardError

Bases: ToolError

Guard denied or crashed during permission check.

exception axio.exceptions.HandlerError

Bases: ToolError

Handler raised during execution.

exception axio.exceptions.StreamError

Bases: AxioError

Error during stream collection.

exception axio.exceptions.ToolError

Bases: AxioError

Base for tool-related errors.

Shared test helpers: StubTransport, fixtures, response builders.

class axio.testing.StubTransport(responses: Sequence[Sequence[StreamEvent | BaseException]] | None = None)

Bases: object

A CompletionTransport that yields pre-configured event sequences.

Each call to stream() pops the next sequence from the list.

stream(messages: list[Message], tools: list[Tool[Any]], system: str) → AsyncIterator[StreamEvent]

axio.testing.assert_static_model_transport_contract(transport: _StaticModelTransport) → None

Check invariants shared by transports with a ready, bundled model catalog.

axio.testing.assert_stream_contract(events: Sequence[StreamEvent]) → None

Check what every CompletionTransport.stream() must produce.

A transport that breaks one of these still passes its own tests, because the agent papers over the difference. Call this from each transport’s tests on whatever its fake server produced.

axio.testing.make_echo_tool() → Tool[Any]

axio.testing.make_ephemeral_context() → MemoryContextStore

axio.testing.make_stub_transport() → StubTransport

axio.testing.make_text_response(text: str = 'Done', iteration: int = 2, usage: Usage | None = None) → list[StreamEvent]

Build a standard end_turn text response event sequence.

axio.testing.make_tool_use_response(tool_name: str = 'echo', tool_id: str = 'call_1', tool_input: dict[str, Any] | None = None, iteration: int = 1, usage: Usage | None = None) → list[StreamEvent]

Build a standard tool_use response event sequence.