axio

Minimal, streaming-first, protocol-driven foundation for LLM-powered agents.

No dependencies. Three protocols. An agent loop that just works.

Features

• Streaming agent loop - run_stream() yields typed events as they arrive; no buffering, no polling
• Three clean protocols - CompletionTransport, ContextStore, PermissionGuard; swap any piece without touching the rest
• Concurrent tool dispatch - all tool calls in a turn run via asyncio.gather automatically
• Context compaction - compact_context() summarises old history to stay within token limits
• Testing helpers - StubTransport, make_tool_use_response(), make_echo_tool() ship in axio.testing
• Plugin-ready - entry-point groups (axio.tools, axio.transport, axio.guards) for drop-in extensions

Installation

pip install axio

Quick start

import asyncio
from axio.agent import Agent
from axio.context import MemoryContextStore
from axio.tool import Tool


# 1. Define a tool
async def greet(name: str) -> str:
    """Return a greeting for the given name."""
    return f"Hello, {name}!"


greet_tool = Tool(name="greet", description="Greet someone by name", handler=greet)

# 2. Wire up the agent (transport comes from an axio-transport-* package)
from axio_transport_openai import OpenAITransport

transport = OpenAITransport(api_key="sk-...", model="gpt-4o-mini")
agent = Agent(system="You are helpful.", transport=transport, tools=[greet_tool])


# 3. Run
async def main() -> None:
    ctx = MemoryContextStore()
    async for event in agent.run_stream("Please greet Alice", ctx):
        print(event)


asyncio.run(main())

Architecture

  User message
       │
       ▼
  ┌─────────┐   stream()    ┌─────────────────────┐
  │  Agent  │ ────────────▶ │ CompletionTransport  │
  │  loop   │ ◀──────────── │ (Anthropic, OpenAI, …) │
  └─────────┘  StreamEvent  └─────────────────────┘
       │
       │ tool_use?
       ▼
  ┌──────────┐   check()   ┌─────────────────┐
  │   Tool   │ ──────────▶ │ PermissionGuard │
  │ handler  │             │ (path, LLM, …)  │
  └──────────┘             └─────────────────┘
       │
       ▼
  ┌──────────────┐
  │ ContextStore │  append() / get_history() / fork() / compact()
  └──────────────┘

Protocols

CompletionTransport

from typing import Protocol, runtime_checkable
from collections.abc import AsyncIterator
from axio.events import StreamEvent
from axio.messages import Message
from axio.tool import Tool


@runtime_checkable
class CompletionTransport(Protocol):
    def stream(self, messages: list[Message], tools: list[Tool], system: str) -> AsyncIterator[StreamEvent]: ...

ContextStore

from axio.context import ContextStore
from axio.messages import Message


class MyContextStore(ContextStore):
    def __init__(self) -> None:
        self._messages: list[Message] = []

    async def append(self, message: Message) -> None:
        self._messages.append(message)

    async def get_history(self) -> list[Message]:
        return list(self._messages)

    # Everything else - session_id, close(), fork(), clear(),
    # get/set_context_tokens(), add_context_tokens(), list_sessions()
    # - has a default implementation.

PermissionGuard

PermissionGuard is an abstract base class (ABC). Subclass it and implement check():

from typing import Any
from axio.permission import PermissionGuard
from axio.tool import Tool


class MyGuard(PermissionGuard):
    async def check(self, tool: Tool[Any], **kwargs: Any) -> dict[str, Any]:
        # return kwargs to allow, raise GuardError to deny
        return kwargs

Stream events

Tools

from axio.tool import Tool


async def summarise(text: str, max_words: int = 20) -> str:
    """Summarise the given text in one sentence."""
    # your implementation
    return "..."


tool = Tool(
    name="summarise",
    description="Summarise text",  # overrides docstring if set
    handler=summarise,
    concurrency=4,  # max parallel executions
)

Testing

from axio.agent import Agent
from axio.testing import (
    StubTransport,
    make_tool_use_response,
    make_text_response,
    make_ephemeral_context,
    make_echo_tool,
)


async def test_agent_calls_tool():
    transport = StubTransport(
        [
            make_tool_use_response("echo", tool_input={"msg": "hi"}),
            make_text_response("Done"),
        ]
    )
    agent = Agent(system="", tools=[make_echo_tool()], transport=transport)
    result = await agent.run("say hi", make_ephemeral_context())
    assert result == "Done"

Plugin entry points

[project.entry-points."axio.tools"]
my_tool = "my_package:MyHandler"

[project.entry-points."axio.transport"]
my_backend = "my_package:MyTransport"

[project.entry-points."axio.guards"]
my_guard = "my_package:MyGuard"

Ecosystem

License

MIT