Appendix A · API Reference

This is a surface reference — the public symbols that form Agentao's embedding contract. Anything not listed here is an implementation detail and may change without notice. For behavior and examples, follow the links back to the main guide.

Authoritative __all__:

• from agentao import ... → Agentao, SkillManager
• from agentao.embedding import ... → build_from_environment
• from agentao.transport import ... → AgentEvent, EventType, CoreTransport, Transport, NullTransport, SdkTransport, NonInteractiveTransport, build_compat_transport, gate_note, record_decline, APPROVAL_UNAVAILABLE
• from agentao.capabilities import ... → FileSystem, LocalFileSystem, FileEntry, FileStat, ShellExecutor, LocalShellExecutor, ShellRequest, ShellResult, BackgroundHandle, MemoryStore, SQLiteMemoryStore, MCPRegistry, FileBackedMCPRegistry, InMemoryMCPRegistry
• from agentao.tools.base import ... → Tool, ToolRegistry (Tool is also exported from agentao.host, with AsyncToolBase and the RegistrableTool union — the same objects, and the stable path for hosts)
• from agentao.permissions import ... → PermissionEngine, PermissionMode, PermissionDecision
• from agentao.memory.manager import MemoryManager
• from agentao.cancellation import ... → CancellationToken, AgentCancelledError (CancellationToken is also exported from agentao.host — the same class, and the stable path for hosts)
• from agentao.acp_client import ... → ACPManager, ACPClient, AcpClientError, AcpErrorCode, AcpRpcError, AcpInteractionRequiredError, AcpClientConfig, AcpServerConfig, AcpConfigError, PromptResult, ServerState, load_acp_client_config (and lower-level re-exports — see agentao.acp_client.__init__.py docstring for which are "stable embedding surface" vs. "implementation detail")
• from agentao.host import ... → ActivePermissions, EventStream, StreamSubscribeError, HostEvent, ToolLifecycleEvent, SubagentLifecycleEvent, SubagentUsage, PermissionDecisionEvent, RFC3339UTCString, export_host_event_json_schema, export_host_acp_json_schema, CancellationToken, Tool, AsyncToolBase, RegistrableTool, TextDelta, TurnOutcome — host-facing harness contract, see A.10

A.1 Agentao

The core class — see Part 2.2 for the full constructor parameter table.

Constructor

See Part 2.2 for the full parameter table. Since 0.3.0, Agentao() without working_directory= raises TypeError from Python signature dispatch — the soft-deprecation cycle ended. Upgrading from 0.2.15: docs/migration/0.2.15-to-0.3.3.md.

Agentao(
    api_key: str | None = None,
    base_url: str | None = None,
    model: str | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None,
    *,                                          # everything below is keyword-only (0.5.0)
    transport: CoreTransport | None = None,
    working_directory: Path,                    # required since 0.3.0
    extra_body: dict | None = None,             # keyword-only; LLM .create() request-body passthrough
    api_format: str | None = None,              # keyword-only; "openai-completions" (default) | "anthropic-messages" | "openai-responses"
    extra_mcp_servers: dict[str, dict] | None = None,
    extra_tools: Sequence[RegistrableTool] | None = None,   # inject / replace tools
    disable_tools: Iterable[str] | None = None,             # skip built-ins by name
    enabled_tools: Iterable[str] | None = None,             # allowlist (mutually excl. w/ disable_tools)
    permission_engine: PermissionEngine | None = None,
    permission_mode: str | None = None,         # "read-only" | "workspace-write" | "full-access"; builds an engine (rules=[])
    max_context_tokens: int = 200_000,
    plan_session: PlanSession | None = None,
    # Embedded-harness explicit injections
    llm_client: LLMClient | None = None,
    logger: logging.Logger | None = None,
    memory_manager: MemoryManager | None = None,
    skill_manager: SkillManager | None = None,
    project_instructions: str | None = None,
    mcp_manager: McpClientManager | None = None,
    mcp_registry: MCPRegistry | None = None,    # 0.3.0+ (#17)
    filesystem: FileSystem | None = None,
    shell: ShellExecutor | None = None,
    # Opt-in subsystems (None = fully disabled)
    replay_config: ReplayConfig | None = None,
    sandbox_policy: SandboxPolicy | None = None,
    bg_store: BackgroundTaskStore | None = None,
    # ...
)
# The 8 legacy callback kwargs (confirmation_callback, step_callback, …) were
# removed in 0.5.0 — passing one is a TypeError. See build_compat_transport below.

Mutual-exclusion rules (raise ValueError if violated):

• llm_client= together with any of api_key= / base_url= / model= / temperature= / max_tokens= / extra_body= / prompt_cache= / prompt_cache_ttl= / api_format= (pass them to the injected client instead)
• mcp_manager= together with extra_mcp_servers=
• mcp_manager= together with mcp_registry= — the registry is a config source, the manager is the construction outcome
• enabled_tools= together with disable_tools= — an allowlist and a denylist at once is ambiguous (holds even for enabled_tools=set()). See 5.1
• permission_engine= together with permission_mode= — one is an engine, the other builds one; set the mode on your engine instead. See 5.4

Opt-in subsystem semantics (defaults are None since 0.2.16):

• replay_config=None — the replay block of <wd>/.agentao/settings.json is not read at construction time and no ReplayManager is attached: agent.replay_manager is None until start_replay() / reload_replay_config() creates one.
• sandbox_policy=None — ToolRunner runs shell commands without the macOS sandbox-exec wrapper.
• bg_store=None — check_background_agent / cancel_background_agent are not registered, the chat loop's background-notification drain short-circuits, and the run_in_background field is schema-level removed from sub-agent tool definitions (the LLM cannot call a disabled feature). /agent bg|dashboard|cancel|delete|logs|result CLI subcommands short-circuit with a clear warning. Leave it None in a one-shot process, as agentao run does (0.5.6): background workers are daemon threads and die with it.
• With a store, a finished background sub-agent's update reaches the parent only on its next turn, and the runtime never starts that turn itself. Either continue from the terminal SubagentLifecycleEvent on your own turn driver (recipe in docs/reference/host-api.md, Continuing after a background sub-agent), or let the model wait in-turn with check_background_agent(agent_id, wait_seconds=…) (0.5.6, at most 1800 s; cancelling the turn ends the wait, not the child).
• A store runs at most max_concurrent background sub-agents at once (keyword-only, default 6; None for no cap; 0.5.7). A launch past it is refused, not queued: register() raises BackgroundCapacityError before anything is recorded, and the tool call fails with text telling the model to wait for one, cancel one, or run in the foreground. Only the store's own pending/running tasks count — not another process's records in the same background_tasks.json.

agentao.embedding.build_from_environment() constructs CLI defaults for all three (anchored to the session's working directory) and passes them explicitly, so CLI / ACP behavior is preserved. Embedded hosts that don't ask for these features pay zero cost.

agentao.embedding.build_from_environment

def build_from_environment(
    working_directory: Path | None = None,
    **overrides,
) -> Agentao:
    ...

CLI-style auto-discovery factory: reads .env, LLM_PROVIDER-prefixed env vars, <wd>/.agentao/settings.json, ~/.agentao/permissions.json (project-scope file is intentionally not loaded — see 5.4), <wd>/.agentao/mcp.json + ~/.agentao/mcp.json (user wins on name collision; project is add-only), memory roots; wires bg_store, sandbox_policy and replay_config (an explicit None disables each); constructs Agentao with the discovered values. Caller-supplied **overrides win over auto-discovered ones. A ~/.agentao/permissions.json that exists but cannot be read or validated raises agentao.embedding.PermissionConfigError (fail closed). Step-by-step order: 2.2 Factory path. Raises ValueError on the same mutual-exclusion conflicts as the constructor (if llm_client is in overrides, the factory's discovered api_key / base_url / model are dropped before forwarding).

Methods

Method	Signature	Purpose
chat	`chat(user_message: str, max_iterations: int = 100, cancellation_token: CancellationToken	None = None, images: list[dict]
arun	`async arun(user_message: str, max_iterations: int = 100, cancellation_token: CancellationToken	None = None, images: list[dict]
astream	`astream(user_message: str, *, max_iterations: int = 100, images: list[dict]	None = None, cancellation_token: CancellationToken
clear_history	clear_history() -> None	Reset self.messages, active skills, todos and token counters; does not touch memory DB. Background agents keep running, but their completion notifications no longer reach the history (read them via check_background_agent).
close	close() -> None	Cancel the background sub-agents this agent launched (waiting up to 5 s; one still running is named in a warning), then release the MCP subprocesses and DB handles the agent opened itself (an injected mcp_manager= / memory_manager= is yours to release). Safe to call more than once. Call in finally:, or use with.
aclose	async aclose() -> None	close() on a thread of its own, so the event loop keeps running. End the agent's turns first; it does not wait for or cancel one.
__enter__ / __exit__, __aenter__ / __aexit__	with Agentao(...) as agent: / async with ... as agent:	Context manager: returns the agent itself; on exit calls close() (sync) or aclose() (async). Exceptions from the block propagate.
set_provider	`set_provider(api_key: str, base_url: str	None = None, model: str
set_model	set_model(model: str) -> str	Swap model only; returns the previous id.
events (0.3.1+)	`events(session_id: str	None = None) -> AsyncGenerator[HostEvent, None]`
active_permissions (0.3.1+)	active_permissions() -> ActivePermissions	Snapshot of the active permission policy (mode, rules, loaded_sources). JSON-safe. See A.10.
add_tool	add_tool(tool: RegistrableTool, *, replace: bool = False) -> None	Register a tool post-construction; same validation + capability binding as extra_tools=. Name clash without replace=True raises (stricter than tools.register). Reserved names (mcp_, plan tools) rejected. Visible next chat()/arun(). See 5.1.
remove_tool	remove_tool(name: str) -> bool	Unregister a tool post-construction; returns whether it existed (absent → False, no raise). mcp_ / plan tools raise. Gone next chat()/arun().

Image input and vision degradation

chat() / arun() accept images=[{"data": <base64>, "mimeType": "image/*"}, …] (0.4.8+). Each image is sent as an OpenAI image_url data-URL part alongside the text. If the selected model rejects image input, the turn retries once with the images degraded to text — one self-closing tag per image, appended at the end of the user message:

<attachment uri="share/screenshot.png" mimetype="image/png"/>

Each item needs a non-empty string data (base64) and mimeType, or the turn raises ValueError before it starts. An optional _source key (origin path or URL) is never sent as image content; it names the image in the tag above and in log labels. The engine checks no size, count or type. The CLI and ACP enforce agentao.media_limits (20 MB per image, 16 images per turn, the bytes sniffed with sniff_image_mime). A host that accepts untrusted images enforces its own caps before calling chat().

uri is the attachment's source path or URL when known (e.g. staged via the CLI /image command); wire images that carry no source degrade to inline-image-N. This section is the canonical description of the format — implementation in agentao/runtime/chat_loop/_runner.py::_render_image_reference_fallback.

Since 0.5.12, when the error is a 400, 413 or 422, the retry also covers images left in history from earlier turns (for example after switching to a text-only model). Each one is replaced with [Image removed from the conversation history because the current model does not accept image input. It can no longer be viewed.]; before, they were sent again and the turn failed again.

Provider-rejected images (0.5.12+). When the provider refuses a request because of an image (a 400, 413 or 422 whose body mentions an image, on a request that carried one, and not only in a run of words echoed from the conversation's user or tool text, a whole line of it or at least five words), every image part in history is replaced with [Image removed from the conversation history after the model provider rejected it. It can no longer be viewed.]. The provider does not say which image it refused, so earlier turns' images go too. Each message keeps its role, text and other parts, and the note never contains the image data. The turn then ends with the [LLM API error: …] notice plus a line asking to re-attach the image, switch model or continue from the text. Without this, the image went out again with every later request and failed the same way, also after a resume. Both rewrites emit IMAGES_REMOVED (replay schema 1.4).

Image URLs a wire cannot send (0.5.12+). A host-written image_url part whose URL the wire cannot express (on anthropic-messages, anything but a base64 data: URL or an http(s) URL; on openai-responses, a missing or empty URL) goes out as [Image omitted: its URL cannot be sent to the model, so it cannot be viewed.], with a warning in agentao.log. History is left as written. Before, the adapter raised before sending, on that turn and every later one. chat(images=...) always builds a data: URL, so it never meets this.

Attributes

Attribute	Type	Notes
messages	list[dict]	Conversation history in OpenAI chat format. Safe to read, mutate at your own risk.
tools	ToolRegistry	The live registry. Prefer the contract APIs — Agentao(extra_tools=[...]) or agent.add_tool(...) — which bind capabilities and validate; agent.tools.register(...) is the low-level path that skips both. See 5.1.
skill_manager	SkillManager	agent.skill_manager.activate_skill(name, task_description) to turn a skill on.
transport	CoreTransport	The active transport; rebindable. With replay enabled it is a ReplayAdapter around the transport you passed. To subscribe, call subscribe() on the transport you constructed, not through this attribute: an isinstance(t, Transport) check passes for a ReplayAdapter whose wrapped transport has no subscribe() and for an explicit Transport subclass's inherited no-op stub, and both register nothing (see 4.1).
_current_token	`CancellationToken	None`

A.2 Transport layer

Transport protocol

Runtime-checkable Protocol. Four methods, all optional via NullTransport fallback:

def emit(self, event: AgentEvent) -> None: ...
def confirm_tool(self, tool_name: str, description: str, args: dict) -> bool: ...
def ask_user(self, question: str) -> str: ...
def on_max_iterations(self, count: int, messages: list) -> dict: ...

on_max_iterations returns {"action": "continue" | "stop" | "new_instruction", "message"?: str}.

SdkTransport

Callback-based adapter — the typical SDK consumer choice:

SdkTransport(
    on_event: Callable[[AgentEvent], None] | None = None,
    confirm_tool: Callable[[str, str, dict], bool] | None = None,
    ask_user: Callable[[str], str] | None = None,
    on_max_iterations: Callable[[int, list], dict] | None = None,
)

Any unset callback falls back to NullTransport behavior (allow, except an MCP Skills gated confirmation, which is refused; empty answer; stop).

build_compat_transport

Helper that builds a Transport from the 8 legacy per-callback args (confirmation_callback, step_callback, thinking_callback, ask_user_callback, output_callback, tool_complete_callback, llm_text_callback, on_max_iterations_callback). Lives in agentao.embedding.compat and is the documented migration surface for hosts still on the pre-0.2.10 callback shape. Since 0.5.0 those kwargs are gone from the Agentao(...) signature, so this is the only way to keep using the callbacks: Agentao(transport=build_compat_transport(confirmation_callback=...), ...). It is not deprecated.

AgentEvent / EventType

@dataclass
class AgentEvent:
    type: EventType
    data: dict

EventType covers UI streaming, tool lifecycle, LLM-call metadata, replay observability, and runtime state changes. Full payload reference in 4.2.

A.3 Tools

Tool (abstract)

class Tool(ABC):
    @property @abstractmethod
    def name(self) -> str: ...
    @property @abstractmethod
    def description(self) -> str: ...
    @property @abstractmethod
    def parameters(self) -> dict: ...       # JSON Schema

    @property
    def requires_confirmation(self) -> bool: return False
    @property
    def is_read_only(self) -> bool: return False
    @property
    def copies_to_subagents(self) -> bool: return False   # opt in per tool; see 5.1

    @abstractmethod
    def execute(self, **kwargs) -> str: ...

    # populated by Agentao at registration time
    working_directory: Path | None
    output_callback: Callable[[str], None] | None
    filesystem: FileSystem | None
    shell: ShellExecutor | None

    def _get_fs(self) -> FileSystem: ...      # lazy LocalFileSystem fallback
    def _get_shell(self) -> ShellExecutor: ... # lazy LocalShellExecutor fallback

Custom tool subclasses that need to read/write files should call self._get_fs() rather than touching pathlib directly — that way the host's injected FileSystem (Docker, virtual FS, audit proxy) is honored.

ToolRegistry

Method	Purpose
register(tool: Tool, *, replace: bool = False, origin: str = "host")	Add a tool. replace=False collision warns and overwrites; replace=True overwrites silently (INFO audit). Low-level — no capability binding; prefer Agentao(extra_tools=) / add_tool (see 5.1). origin records where the tool came from (builtin / host / mcp / agent / plan); hosts leave it at host. A replacement overwrites it
origin(name: str) -> str	The recorded origin. Sub-agents decide by it: a host tool reaches a sub-agent only when the tool declares copies_to_subagents, and then as a copy — including a host tool that replaced a built-in with an instance of the built-in's own class, which is why class identity is not the test. A tool put in tools without register reads as host. Raises KeyError for an unknown name
unregister(name: str) -> bool	Remove name; returns whether it existed. Pure dict op
get(name: str) -> Tool	Raises KeyError with available-tools list
list_tools() -> list[Tool]
to_openai_format() -> list[dict]	OpenAI function-calling schemas

Capabilities (agentao.capabilities)

Runtime-checkable Protocols that file/search/shell tools route IO through. Hosts inject custom implementations via Agentao(filesystem=..., shell=...) to redirect through Docker exec, virtual filesystems, audit proxies, or remote runners. The package also exports byte-equivalent LocalFileSystem / LocalShellExecutor defaults.

class FileSystem(Protocol):
    def read_bytes(self, path: Path) -> bytes: ...
    def read_partial(self, path: Path, n: int) -> bytes: ...
    def open_text(self, path: Path) -> Iterator[str]: ...      # streaming
    def write_text(self, path: Path, data: str, *, append: bool = False) -> None: ...
                                                               # replacing existing
                                                               # content MUST be atomic
    def list_dir(self, path: Path) -> list[FileEntry]: ...
    def glob(self, base: Path, pattern: str, *, recursive: bool) -> list[Path]: ...
    def stat(self, path: Path) -> FileStat: ...
    def exists(self, path: Path) -> bool: ...
    def is_dir(self, path: Path) -> bool: ...
    def is_file(self, path: Path) -> bool: ...

class ShellExecutor(Protocol):
    def run(self, request: ShellRequest) -> ShellResult: ...
    def run_background(self, request: ShellRequest) -> BackgroundHandle: ...

Frozen dataclasses: FileEntry(name, is_dir, is_file, size), FileStat(size, mtime, is_dir, is_file), ShellRequest(launch, timeout, on_chunk, cancellation_token), ShellResult(returncode, stdout, stderr, timed_out, stdout_omitted_bytes, stderr_omitted_bytes, stdout_omitted_at, stderr_omitted_at, cancelled), BackgroundHandle(pid, pgid, command, cwd).

ShellResult.stdout_omitted_bytes / stderr_omitted_bytes (default 0) count the bytes of a stream that were not kept, and stdout_omitted_at / stderr_omitted_at (default 0) give the byte offset in stdout / stderr where they were — 0 meaning the front, so a tail-only executor sets only the count. LocalShellExecutor keeps the first and last 512 KiB of each stream; a custom executor that keeps everything leaves all four at 0. ShellRequest.cancellation_token (default None) is the turn's CancellationToken: LocalShellExecutor kills the process tree when it fires and returns ShellResult.cancelled=True (default False); an executor that ignores it runs the command to its end or timeout, as before.

A ShellRequest carries a launch, not a command (changed in 0.4.22; it used to hold command, cwd and env directly). request.launch is one of two shapes, and an executor picks them apart with isinstance:

Shape	Fields	How to run it
LegacyLaunch	command, cwd, env, executable	subprocess.Popen(command, shell=True, executable=executable or None, cwd=cwd, env=env) — what every host got before, and still gets unless an interpreter is configured
WindowsLaunch	application_name, command_line, cwd, env	start application_name with command_line and no shell in between; produced when a shell block names an interpreter

request.command and request.cwd survive as read-only properties, so an executor that only logs or audits them needs no change; request.env is gone — read it from the launch. An executor may additionally implement ShellSpecProvider, a shell_spec property answering ShellSpec | Exhausted, to declare which interpreter it will run; the command floor and the prompt's shell guidance both read that answer. Full contract: docs/reference/host-api.md.

Hosts that cannot support real backgrounding may raise NotImplementedError from run_background — ShellTool surfaces that as a normal tool error string.

write_text carries a contract obligation beyond its signature: an implementation that replaces the content of an existing file must do so atomically, so an interrupted write (Ctrl+C, OOM kill) can never leave the user's file truncated. append=True and not-yet-existing targets keep the direct-write path — neither can destroy existing content. LocalFileSystem is the reference: it stages a sibling .{name}.*.tmp in the target's directory (visible to audit/watcher wrappers) and os.replaces it, and it raises PermissionError on a read-only target rather than letting directory-level rename permission silently defeat chmod 444.

A.4 Permissions

PermissionMode

class PermissionMode(Enum):
    READ_ONLY = "read-only"
    WORKSPACE_WRITE = "workspace-write"
    FULL_ACCESS = "full-access"
    PLAN = "plan"

PermissionDecision

class PermissionDecision(Enum):
    ALLOW = "allow"
    DENY = "deny"
    ASK = "ask"

PermissionEngine

PermissionEngine(
    project_root: Path,
    *,
    user_root: Path | None = None,
)

Both arguments are explicit since 0.2.16 — passing project_root=None (or omitting it) raises TypeError. User rules load from <user_root>/permissions.json (typically ~/.agentao/permissions.json if user_root=Path.home() / ".agentao"); pass user_root=None to disable user-scope rules entirely. project_root is used only to detect (and warn on) a stray <project_root>/.agentao/permissions.json — that file is not loaded as a rule source (see 5.4 for why). Switch the preset mode after construction with set_mode().

Method	Signature	Purpose
decide	`decide(tool_name: str, tool_args: dict) -> PermissionDecision	None`
set_mode	set_mode(mode: PermissionMode) -> None	Switch active preset (READ_ONLY, WORKSPACE_WRITE, FULL_ACCESS, PLAN)
active_mode	attribute (read)	Currently active PermissionMode

Subclass and override decide() to integrate company IAM — see 7.3 for a confidence-gated example.

A.5 Memory

MemoryManager (since 0.3.0 / #16)

MemoryManager(
    project_store: MemoryStore,                 # required, pre-built
    user_store: MemoryStore | None = None,      # cross-project; None disables user scope
    guard: MemoryGuard | None = None,
)

The constructor takes pre-built MemoryStore instances (the embedding factory builds them and passes them in). Path-based construction moved to the store layer:

from agentao.memory import MemoryManager, SQLiteMemoryStore

mgr = MemoryManager(
    project_store=SQLiteMemoryStore.open_or_memory(workdir / ".agentao" / "memory.db"),
    user_store=SQLiteMemoryStore.open(home / ".agentao" / "memory.db"),
)

• project_store — required MemoryStore for project-scoped memories (the factory uses SQLiteMemoryStore.open_or_memory(...) so a missing directory degrades to :memory: instead of crashing)
• user_store — optional cross-project store (the factory uses SQLiteMemoryStore.open(...); None disables user-scope memory entirely; user-scope writes are downgraded to project on None)
• guard — optional MemoryGuard that filters sensitive content before persistence

Method	Purpose
upsert(req)	Insert / update a memory record
save_from_tool(...)	Called by the save_memory tool
get_entry(id) / get_all_entries(scope=?)	Read entries
search(query, scope=?) / filter_by_tag(tag, scope=?)	Search
delete(id) / delete_by_title(title)	Soft delete
clear(scope=?)	Soft-delete all in scope
save_session_summary(...) / get_recent_session_summaries(...)	Used by the compaction pipeline
archive_session(session_id=None) / clear_session()	End-of-session house-keeping. Pass the conversation's id to archive_session — summaries are written under it and the cross-session tail excludes only it, so a resumed session must adopt its own id; no argument gives a fresh random id
clear_all_session_summaries()	Drop every session summary across all sessions
wipe_all()	Hard reset: memories (both scopes, soft delete — rows stay in the file) + every session summary. Never raises — returns a MemoryWipeResult whose ok is the success signal. Does not clear the review queue
get_stable_entries(...)	Render into <memory-stable> system-prompt block

MemoryStore (Protocol, 0.3.0 / #16)

class MemoryStore(Protocol):
    # Memory CRUD
    def upsert_memory(self, record: MemoryRecord) -> MemoryRecord: ...
    def get_memory_by_id(self, memory_id: str) -> Optional[MemoryRecord]: ...
    def get_memory_by_scope_key(self, scope: str, key_normalized: str) -> Optional[MemoryRecord]: ...
    def list_memories(self, scope: Optional[str] = None) -> List[MemoryRecord]: ...
    def search_memories(self, query: str, scope: Optional[str] = None) -> List[MemoryRecord]: ...
    def filter_by_tag(self, tag: str, scope: Optional[str] = None) -> List[MemoryRecord]: ...
    def soft_delete_memory(self, memory_id: str) -> bool: ...
    def clear_memories(self, scope: Optional[str] = None) -> int: ...
    # Session summaries
    def save_session_summary(self, record: SessionSummaryRecord) -> None: ...
    def list_session_summaries(self, session_id=None, limit=20) -> List[SessionSummaryRecord]: ...
    def clear_session_summaries(self, session_id=None) -> int: ...
    # Review queue
    def upsert_review_item(self, item: MemoryReviewItem) -> MemoryReviewItem: ...
    def get_review_item(self, item_id: str) -> Optional[MemoryReviewItem]: ...
    def list_review_items(self, status="pending", limit=50) -> List[MemoryReviewItem]: ...
    def update_review_status(self, item_id: str, status: str) -> bool: ...

15 methods, schema-less. Implement the Protocol to back memory with Redis / Postgres / in-process dict / remote API. Default SQLiteMemoryStore is byte-equivalent to the pre-Protocol implementation.

Optional, outside the Protocol: change_version() -> int, a counter that advances on every write to the store's memories by any writer (another manager, another process). MemoryRetriever uses it to refresh its recall index after such writes. A store without it (or whose answer is not an int) still works; recall then sees only writes made through its own MemoryManager. SQLiteMemoryStore implements it with triggers in the database file.

SQLiteMemoryStore

SQLiteMemoryStore.open(path)              # strict; raises on disk error
SQLiteMemoryStore.open_or_memory(path)    # graceful; degrades to ":memory:"

The classmethods replace the historical MemoryManager.__init__ try/except. Use open_or_memory for the project store (a missing DB beats a crashed agent), open for the user store (a failure should disable the scope, not silently re-route writes).

A.6 MCP

MCPRegistry (Protocol, 0.3.0 / #17)

class MCPRegistry(Protocol):
    def list_servers(self) -> Dict[str, McpServerConfig]: ...

Source of MCP server configs. The Agentao(mcp_registry=...) kwarg is the injection point; mcp_manager= and mcp_registry= are mutually exclusive (manager = construction outcome, registry = config source).

Concrete class	Purpose
FileBackedMCPRegistry(project_root, user_root=None)	CLI/ACP default. Reads <wd>/.agentao/mcp.json + <user_root>/mcp.json on every list_servers(). User wins on name collision; project entries may only declare new server names (collisions log a warning and skip the project entry).
InMemoryMCPRegistry(servers=None)	Programmatic counterpart for tests / embedded hosts. Constructor input is shallow-copied.

A.7 Cancellation

CancellationToken

token = CancellationToken()
token.is_cancelled         # bool, non-throwing
token.cancel("reason")     # idempotent
token.check()              # raises AgentCancelledError if cancelled
token.reason               # str

Pass to agent.chat(..., cancellation_token=token) to abort from another thread / async task. See 2.3 for the timeout pattern.

AgentCancelledError

Raised inside the agent loop when a token is cancelled. Caught by chat(), which returns a [Cancelled: reason] string rather than propagating.

A.8 ACP client

ACPManager

Typical host-side entry point for driving external ACP servers declared in .agentao/acp.json. In unattended environments (CI / workers / cron / queue consumers), this same object is what the guide calls the Headless Runtime: not a new protocol or class, just an operating profile of ACPManager.

Method	Signature	Purpose
from_project	@classmethod from_project(project_root=None, *, terminal_auth=False) -> ACPManager	Read .agentao/acp.json. terminal_auth=True declares ACP Terminal Auth — only when the host can run a terminal login
add_server(name, config)		Register a new server with a running manager, stopped; running servers untouched
auth_methods(name)	-> list[dict]	authMethods from the server's last initialize
needs_login(name)	-> bool	The server's last session setup answered auth_required; stays true across restarts until a session opens or setup fails for another reason
reserve_for_login(name)	context manager	Keeps turns / connects from other threads off name (SERVER_BUSY) during a login → restart → connect
server_names	-> list[str]	Declared servers
start_all	start_all(only_auto=True)	Spawn all auto-start servers
start_server(name) / stop_server(name) / restart_server(name)
ensure_connected(name, cwd=?, mcp_servers=?)		Idempotent connect + session
send_prompt(name, text, timeout=?, interactive=True, cwd=?, mcp_servers=?, interaction_policy=?)	-> dict (the session/prompt result)	Turn on a long-lived session. Fail-fast: raises SERVER_BUSY if a turn is active (public)
prompt_once(name, prompt, cwd=?, mcp_servers=?, timeout=?, interactive=False, stop_process=True)	-> PromptResult	Fail-fast one-shot, cleans up (public)
send_prompt_nonblocking / finish_prompt_nonblocking / cancel_prompt_nonblocking		Lower-level async variants (internal / unstable — not part of the embedding contract)
stop_all()		Shut down all subprocesses
get_status()	-> list[ServerStatus]	Typed headless snapshot (Week 1 frozen + Week 2 additive fields)
readiness(name)	-> Literal["ready","busy","failed","not_ready"]	Typed classification of state × active-turn (Week 2)
is_ready(name)	-> bool	Shortcut for readiness(name) == "ready"
reset_last_error(name)		Clear recorded last_error / last_error_at on the manager
get_client(name) / get_handle(name)		Introspection
config (property)	-> AcpClientConfig

PromptResult

@dataclass
class PromptResult:
    stop_reason: str
    raw: Any
    session_id: str | None
    cwd: str | None

ServerState

class ServerState(str, Enum):
    CONFIGURED = "configured"
    STARTING = "starting"
    INITIALIZING = "initializing"
    READY = "ready"
    BUSY = "busy"
    WAITING_FOR_USER = "waiting_for_user"
    STOPPING = "stopping"
    STOPPED = "stopped"
    FAILED = "failed"

ServerStatus

Typed return type of ACPManager.get_status(). Week 1 core fields are frozen; Week 2 added diagnostic fields additively.

@dataclass(frozen=True)
class ServerStatus:
    # Week 1 — core (frozen)
    server: str
    state: str
    pid: Optional[int]
    has_active_turn: bool

    # Week 2 — diagnostics (additive)
    active_session_id: Optional[str] = None
    last_error: Optional[str] = None
    last_error_at: Optional[datetime] = None   # tz-aware, UTC
    inbox_pending: int = 0
    interaction_pending: int = 0
    config_warnings: List[str] = field(default_factory=list)

Week 2 field notes:

• last_error is sticky across successful turns; clear with reset_last_error(name).
• last_error_at is assigned at store time (inside the manager), not at raise time. Use it for staleness judgements.
• SERVER_BUSY and SERVER_NOT_FOUND are not recorded — they are caller-side signals.

See 3.4 Status fields and Readiness classes for the exact field meanings and the readiness classifier, and Appendix D.5 for the state-vs-error contract.

Exception classes

Class	code	Notes
AcpClientError	any AcpErrorCode	Base
AcpServerNotFound	SERVER_NOT_FOUND	Also inherits KeyError
AcpRpcError	.rpc_code: int + .error_code: PROTOCOL_ERROR	JSON-RPC response error
AcpInteractionRequiredError	INTERACTION_REQUIRED	Carries prompt + options

See Appendix D for the full code table.

load_acp_client_config

load_acp_client_config(project_root: Path | None = None) -> AcpClientConfig

When project_root is None, it defaults to Path.cwd().

Validate + load .agentao/acp.json without constructing a manager. Useful for config-lint tooling.

A.9 Skills

SkillManager

Loaded lazily via from agentao import SkillManager. Most callers reach it via agent.skill_manager.

Method	Purpose
list_available_skills() -> list[str]	Names currently discoverable
list_all_skills() -> list[str]	Includes disabled
`get_skill_info(name) -> dict	None`
activate_skill(name, task_description)	Turn on — injects SKILL.md + active reference files into system prompt
enable_skill(name) / disable_skill(name)	Persistent enable/disable in config

A.10 Embedded Host Contract

Stable since 0.3.1. Package renamed agentao.harness → agentao.host in 0.4.2.

The agentao.host package is the stable host-facing API surface for embedding Agentao. Internal runtime types (AgentEvent, ToolExecutionResult, PermissionEngine) are intentionally not part of this surface — they may change in any release. Hosts that target only agentao.host (plus the Agentao(...) constructor and the methods marked above) stay forward-compatible.

Naming. "Harness" still refers to Agentao itself running embedded in a host application (the conceptual framing in docs/design/embedded-host-contract.md); the contract package was renamed to agentao.host to make from agentao.host import HostEvent read consistently. The old agentao.harness import path and old symbol names (HarnessEvent, HarnessReplaySink, export_harness_*) warned from 0.4.2 and were removed in 0.5.0 — replace harness / Harness with host / Host.

Full reference: docs/reference/host-api.md. Design rationale: docs/design/embedded-host-contract.md.

Public exports

from agentao.host import (
    ActivePermissions,
    EventStream,
    StreamSubscribeError,
    HostEvent,
    ToolLifecycleEvent,
    SubagentLifecycleEvent,
    SubagentUsage,
    PermissionDecisionEvent,
    RFC3339UTCString,
    export_host_event_json_schema,
    export_host_acp_json_schema,
    CancellationToken,
    Tool,
    AsyncToolBase,
    RegistrableTool,
    TextDelta,
    TurnOutcome,
)

Symbol	Purpose
ActivePermissions	Read-only snapshot of the active permission policy (mode, rules, loaded_sources).
ToolLifecycleEvent	Public envelope for one tool call. phase ∈ {started, completed, failed}; cancellation surfaces as phase="failed", outcome="cancelled".
SubagentLifecycleEvent	Lineage fact for a sub-agent task/session. phase ∈ {spawned, completed, failed, cancelled} — cancelled is a distinct phase here. failed means either "the sub-agent raised" (error_type = exception class name) or "it returned without ever answering" (error_type = "incomplete:<reason>", <reason> ∈ no_output, reasoning_only, length_truncated, doom_loop, llm_error, max_iterations). Branch on error_type before escalating — see 4.7.
SubagentUsage	The token counts on a terminal SubagentLifecycleEvent.usage (0.5.2): prompt_tokens (the whole input), completion_tokens, cache_read_tokens, cache_creation_tokens — the last two are parts of prompt_tokens — and, since 0.5.15, reasoning_tokens, a part of completion_tokens (0 on a wire that does not state it). None on spawned and wherever no sub-agent was built. Already included in the parent's session totals when the event fires.
PermissionDecisionEvent	Per-decision projection. Fires on allow / deny / prompt; consumers must drain even allow events.
HostEvent	Discriminated union of the three event models (Pydantic discriminator: event_type).
RFC3339UTCString	Constrained timestamp type. Canonical Z suffix only — +00:00 offsets are rejected.
EventStream	Runtime side of Agentao.events(). Reference only: Agentao keeps its stream private, so a host reads it through agent.events() or agent.add_host_event_observer(), never by calling publish() / subscribe() itself.
StreamSubscribeError	Raised when a second concurrent subscriber for the same session_id filter is iterated while the first is open — raised on its first iteration, not when agent.events() is called (MVP supports one consumer per filter; close the first with aclosing, or fan out with synchronous observers).
export_host_event_json_schema()	Emit the canonical JSON schema for events + permissions. Used by tests/test_host_schema.py for byte-equality against docs/schema/host.events.v2.json.
export_host_acp_json_schema()	Emit the canonical JSON schema for host-facing ACP payloads. Snapshot lives at docs/schema/host.acp.v1.json.
CancellationToken	The token chat() / arun() accept as cancellation_token=. Re-export of agentao.cancellation.CancellationToken (the same class).
Tool, AsyncToolBase	Base classes for host-supplied tools (extra_tools= / add_tool). Re-export of the canonical types in agentao.tools.base.
RegistrableTool	Union[Tool, AsyncToolBase] — what extra_tools= / add_tool accept.
TextDelta	One chunk of assistant text from Agentao.astream() (text: str). For display: joined deltas are not the answer. Not a HostEvent member; not projected into replay or the schema snapshot.
TurnOutcome	How a turn ended: text, status, incomplete_reason, tool_count, error, finish_reason_missing, and .is_answer. The last item of astream(), and what agent.last_turn returns. Same class as agentao.TurnOutcome.

agent.events(session_id=None)

Returns an async iterator over HostEvent. Pass session_id= to filter; pass None to subscribe to every session owned by this Agentao instance.

async for ev in agent.events():
    if isinstance(ev, ToolLifecycleEvent):
        ...
    elif isinstance(ev, PermissionDecisionEvent):
        ...
    elif isinstance(ev, SubagentLifecycleEvent):
        ...

Delivery contract:

• Same-session ordering is guaranteed; cross-session global ordering is not.
• Within one tool_call_id, PermissionDecisionEvent precedes ToolLifecycleEvent(phase="started").
• No replay. Events emitted before the first subscription are dropped.
• Backpressure is host-pulled via a bounded queue — slow consumers block the producer for matching events rather than dropping them.
• Cancelling the iterator releases queue/subscription resources.
• MVP supports one async-iterator consumer per session_id filter; a second one on the same filter raises StreamSubscribeError on its first iteration while the first is open. For several consumers, use agent.add_host_event_observer().

agent.active_permissions()

Returns a JSON-safe ActivePermissions snapshot:

snap = agent.active_permissions()
# snap.mode            -> "workspace-write" (Literal-typed)
# snap.rules           -> list[dict]
# snap.loaded_sources  -> ["preset:workspace-write",
#                          "user:/Users/me/.agentao/permissions.json",
#                          "injected:host"]

loaded_sources carries stable string labels: preset:<mode>, user:<path>, injected:<name>. (project:<path> is no longer emitted — project-scope permissions are not loaded; see 5.4.) The MVP does not expose per-rule provenance — hosts that need it combine loaded_sources with their own injected policy metadata.

If no permission_engine is configured, the runtime returns a permissive fallback: mode="workspace-write", empty rules, loaded_sources=["default:no-engine"]. The label tells hosts they're seeing the engine-less fallback rather than a configured policy: no rule is evaluated, not even the workspace-write preset's, and every ask goes to the transport (which NullTransport approves).

Hosts that layer policy on top of the engine call agent.permission_engine.add_loaded_source("injected:<name>") so the snapshot reflects their provenance. The cache is invalidated on set_mode() and add_loaded_source().

Schema snapshots

Each release ships checked-in JSON schema snapshots:

• docs/schema/host.events.v2.json — events + permissions surface
• docs/schema/host.acp.v1.json — host-facing ACP payloads

tests/test_host_schema.py regenerates the schemas from the Pydantic models and asserts byte-equality. A model change that shifts the wire form must update both the model and the snapshot in the same PR. Adding an optional field is backwards-compatible; removing or renaming requires a schema version bump.

Non-goals (explicitly outside the contract)

• Public agent graph / descendants store API
• Host-facing hooks list/disable API (rule-author guide: §5.7 Plugin Hooks)
• Host-facing MCP reload / lifecycle events
• Local plugin export/import; remote plugin share
• External session import
• Generated client SDKs

The CLI may build on the same events for its own UI, but its stores and commands are not promoted to the host contract.

---

→ Appendix F · FAQ