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)

Bases: object

copy(**overrides: Any) → Self

Return a new Agent with overrides applied.

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

last_iteration_message: Message | None

max_iterations: int

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

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

selector: ToolSelector | None

system: str

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

Bases: Generic

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

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)

Bases: object

copy(**overrides: Any) → Self

Return a new Agent with overrides applied.

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

last_iteration_message: Message | None

max_iterations: int

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

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

selector: ToolSelector | None

system: str

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

Bases: Generic

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

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.

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 protocols: completion, image gen, TTS, STT.

Transports should be stateless - all request state lives in the arguments passed to each method call. This allows multiple agents to share a single transport instance and call it concurrently without interference.

The one allowed exception is a reusable connection pool (e.g. an aiohttp.ClientSession), which is safe to share across concurrent calls.

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.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.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.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]

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_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.

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_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.