附录 A · API 参考

这是一份面向表层的参考——Agentao 嵌入契约里的公开符号。未在此列出的都算实现细节，可能变动。行为与示例请回跳到正文相应章节。

权威 __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 也从 agentao.host 导出，同时导出的还有 AsyncToolBase 和联合类型 RegistrableTool——是同一批对象，也是宿主应走的稳定路径）
• from agentao.permissions import ... → PermissionEngine、PermissionMode、PermissionDecision
• from agentao.memory.manager import MemoryManager
• from agentao.cancellation import ... → CancellationToken、AgentCancelledError（CancellationToken 也从 agentao.host 导出——同一个类，是宿主应使用的稳定路径）
• from agentao.acp_client import ... → ACPManager、ACPClient、AcpClientError、AcpErrorCode、AcpRpcError、AcpInteractionRequiredError、AcpClientConfig、AcpServerConfig、AcpConfigError、PromptResult、ServerState、load_acp_client_config（以及更底层的 re-export——哪些属于"稳定嵌入面"、哪些属于"实现细节"请参考 agentao.acp_client.__init__.py 的 docstring）
• 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 —— 宿主面 harness 合约，详见 A.10

A.1 Agentao

核心类——完整构造参数表见 Part 2.2。

构造器

完整参数表见 Part 2.2。0.3.0 起，不传 working_directory= 调用 Agentao() 会从 Python 签名分派直接抛 TypeError——软废弃周期已结束。从 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,
    *,                                           # 以下全部仅限关键字（0.5.0）
    transport: CoreTransport | None = None,
    working_directory: Path,                     # 0.3.0 起必传
    extra_body: dict | None = None,              # 仅关键字；LLM .create() 请求体直通
    api_format: str | None = None,               # 仅关键字；"openai-completions"（默认）| "anthropic-messages" | "openai-responses"
    extra_mcp_servers: dict[str, dict] | None = None,
    extra_tools: Sequence[RegistrableTool] | None = None,   # 注入 / 替换工具
    disable_tools: Iterable[str] | None = None,             # 按名跳过内置
    enabled_tools: Iterable[str] | None = None,             # 白名单（与 disable_tools 互斥）
    permission_engine: PermissionEngine | None = None,
    permission_mode: str | None = None,         # "read-only" | "workspace-write" | "full-access"；构造引擎（rules=[]）
    max_context_tokens: int = 200_000,
    plan_session: PlanSession | None = None,
    # 嵌入式 harness 显式注入
    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,
    # 可选启用的子系统（None = 完全禁用）
    replay_config: ReplayConfig | None = None,
    sandbox_policy: SandboxPolicy | None = None,
    bg_store: BackgroundTaskStore | None = None,
    # ...
)
# 8 个 legacy 回调 kwarg（confirmation_callback、step_callback……）已于 0.5.0
# 移除 —— 传入任何一个都是 TypeError。见下文 build_compat_transport。

互斥规则（违反时抛 ValueError）：

• llm_client= 与任何 api_key= / base_url= / model= / temperature= / max_tokens= / extra_body= / prompt_cache= / prompt_cache_ttl= / api_format= 同时传（改为把它们传给注入的 client）
• mcp_manager= 与 extra_mcp_servers= 同时传
• mcp_manager= 与 mcp_registry= 同时传——registry 是配置源，manager 是构造结果
• enabled_tools= 与 disable_tools= 同时传——白名单和黑名单并存有歧义（enabled_tools=set() 也照样报错）。见 5.1
• permission_engine= 与 permission_mode= 同时传——一个是引擎，一个会构造引擎；请在你自己的引擎上设置模式。见 5.4

可选子系统语义（默认 None）：

• replay_config=None —— 构造时不读 <wd>/.agentao/settings.json 的 replay 块，也不挂 ReplayManager：在 start_replay() / reload_replay_config() 创建它之前，agent.replay_manager 是 None。
• sandbox_policy=None —— ToolRunner 跑 shell 时不再走 macOS sandbox-exec 包装。
• bg_store=None —— check_background_agent / cancel_background_agent 不注册，chat loop 后台通知 drain 短路，子 agent 工具定义里 run_in_background 字段在 schema 层被移除（LLM 看不到、就不会调用一个被禁用的能力）。/agent bg|dashboard|cancel|delete|logs|result 这几个 CLI 子命令也会短路，并打印明确的提示。一次性进程应保持 None，agentao run 就是这样做的（0.5.6）：后台 worker 是 daemon 线程，会随进程一起结束。
• 有 store 时，后台子代理完成后的更新只会在父级下一轮交给它，而运行时从不自己开启这一轮。两种做法任选：在宿主自己驱动轮次的地方，根据终态 SubagentLifecycleEvent 续跑（配方见 docs/reference/host-api.zh.md《后台子 Agent 结束后继续》）；或者让模型用 check_background_agent(agent_id, wait_seconds=…) 在轮次内等待（0.5.6，最多 1800 秒；取消这一轮只结束等待，不取消子代理）。
• 一个 store 同时最多运行 max_concurrent 个后台子代理（keyword-only，默认 6；None 表示不设上限；0.5.7）。超出时直接拒绝，不排队：register() 在记录任何内容之前抛出 BackgroundCapacityError，这次工具调用以失败返回，文字告诉模型等一个结束、取消一个，或改为前台运行。只统计这个 store 自己 pending/running 的任务，不算同一个 background_tasks.json 里其他进程的记录。

agentao.embedding.build_from_environment() 会按 CLI 默认行为构造这三个对象（都锚定到当前 session 的工作目录），然后显式传入，所以 CLI / ACP 行为保持不变。嵌入式 host 不主动启用就不会有任何开销。

agentao.embedding.build_from_environment

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

CLI 风格的自动发现工厂：读 .env、LLM_PROVIDER 前缀的 env 变量、<wd>/.agentao/settings.json、~/.agentao/permissions.json（项目级文件故意不加载 —— 见 5.4）、<wd>/.agentao/mcp.json + ~/.agentao/mcp.json（同名冲突时用户胜；项目级仅可新增）、内存目录；接上 bg_store、sandbox_policy、replay_config（显式传 None 即禁用）；然后用发现到的值构造 Agentao。调用方传入的 **overrides 优先。~/.agentao/permissions.json 存在但读不了或校验不过时，抛 agentao.embedding.PermissionConfigError（失败即关闭）。逐步顺序见 2.2 工厂路径。同样会触发上面构造器的互斥校验（如果 overrides 含 llm_client，工厂会先把发现到的 api_key / base_url / model 丢掉再转发）。

方法

方法	签名	作用
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	清 self.messages、已激活 skills、todos 与 token 计数；不影响 memory DB。后台 agent 继续运行，但完成通知不再进入历史（用 check_background_agent 查看）
close	close() -> None	先取消这个 agent 启动的后台子代理（最多等 5 秒；仍在运行的会在 warning 里写明），再关 agent 自己打开的 MCP 子进程与 DB handle（注入的 mcp_manager= / memory_manager= 由你释放）；可重复调用。请放 finally:，或用 with
aclose	async aclose() -> None	在独立线程上运行 close()，不阻塞事件循环。先结束 agent 的轮次；它不会等待或取消正在进行的轮次
__enter__ / __exit__、__aenter__ / __aexit__	with Agentao(...) as agent: / async with ... as agent:	context manager：返回 agent 本身；退出时调用 close()（同步）或 aclose()（异步）。块内的异常照常抛出
set_provider	set_provider(api_key, base_url=None, model=None, *, api_format=None) -> None	运行时换 LLM。api_format（0.5.0）指明新 provider 的线路协议；None 保持当前线路
set_model	set_model(model: str) -> str	只换模型；返回旧 id
events (0.3.1+)	`events(session_id: str	None = None) -> AsyncGenerator[HostEvent, None]`
active_permissions (0.3.1+)	active_permissions() -> ActivePermissions	当前权限策略快照（mode、rules、loaded_sources），JSON-safe。详见 A.10
add_tool	add_tool(tool: RegistrableTool, *, replace: bool = False) -> None	构造后注册工具；与 extra_tools= 同样的校验 + 能力绑定。名字冲突且未传 replace=True 时抛异常（比 tools.register 更严）。保留名（mcp_、plan 工具）拒绝。下一次 chat()/arun() 可见。见 5.1
remove_tool	remove_tool(name: str) -> bool	构造后移除工具；返回是否存在过（缺失 → False，不抛）。mcp_ / plan 工具会抛异常。下一次 chat()/arun() 消失

图片输入与视觉退化

chat() / arun() 接受 images=[{"data": <base64>, "mimeType": "image/*"}, …] （0.4.8+）。每张图片以 OpenAI image_url data-URL part 随文本一起送入模型。 模型拒绝图片输入时，本轮自动重试一次，把图片退化为文本——每张图一个自闭合标签， 附在 user message 末尾：

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

每一项都要有非空字符串 data（base64）和 mimeType，否则本轮开始前就抛 ValueError。可选的 _source 键（来源路径或 URL）从不作为图片内容发送，只用于上面标签里的命名和日志标签。引擎不检查大小、数量或类型。 CLI 和 ACP 执行 agentao.media_limits 的上限（每张 20 MB，每轮 16 张，并用 sniff_image_mime 嗅探字节）。接收不可信图片的宿主，要在调用 chat() 前自己设上限。

uri 取已知的来源路径或 URL（例如经 CLI /image 命令暂存的文件）；wire 上不带 来源的图片退化为 inline-image-N。本节是该格式的规范出处——实现见 agentao/runtime/chat_loop/_runner.py::_render_image_reference_fallback。

自 0.5.12 起，当报错状态码为 400、413 或 422 时，这次重试也覆盖历史里之前轮次留下的图片（例如切换到纯文本模型之后）。 每张都替换为 [Image removed from the conversation history because the current model does not accept image input. It can no longer be viewed.]；此前它们会被再次发送，这一轮也会再次失败。

服务商拒收的图片（0.5.12+）。服务商因为某张图片拒绝请求时（状态码 400、413 或 422、报错正文提到图片、且请求里带了图片；若正文里每个“image”都只出现在回显对话里用户或工具原文的一串词里（整行原文或至少五个词），则不算），历史里所有图片部分都替换为 [Image removed from the conversation history after the model provider rejected it. It can no longer be viewed.]。服务商不会指明拒收的是哪一张，所以之前轮次的图片也一并移除。每条消息 保留其角色、文字和其他部分，说明文字里绝不包含图片数据。随后这一轮以 [LLM API error: …] 提示结束，并附一句话请用户重新附图、换模型或根据文字继续。 没有这一步，这张图片会随之后每次请求再次发送并以同样方式失败，恢复会话后也一样。 两种改写都会发出 IMAGES_REMOVED（replay schema 1.4）。

线路无法发送的图片 URL（0.5.12+）。宿主写入的 image_url 部分，如果其 URL 是当前线路无法表达的（anthropic-messages 上除 base64 data: URL 和 http(s) URL 以外的任何形式；openai-responses 上缺失或为空的 URL），发出的请求里会以 [Image omitted: its URL cannot be sent to the model, so it cannot be viewed.] 代替，并在 agentao.log 里记一条警告。历史保持原样。此前适配器会在发送前抛错， 这一轮和之后每一轮都如此。chat(images=...) 总是生成 data: URL，不会遇到这种情况。

属性

属性	类型	说明
messages	list[dict]	OpenAI chat 格式的历史；可读，修改风险自担
tools	ToolRegistry	活的注册表。优先用契约 API——Agentao(extra_tools=[...]) 或 agent.add_tool(...)（会绑定能力并校验）；agent.tools.register(...) 是绕过两者的底层路径。见 5.1
skill_manager	SkillManager	agent.skill_manager.activate_skill(name, task_description) 激活技能
transport	CoreTransport	当前传输层；可重新赋值。开启 replay 时，它是包着你传入的 transport 的 ReplayAdapter。要订阅，就在你自己构造的 transport 上调用 subscribe()，不要通过这个属性：isinstance(t, Transport) 对包装的 transport 没有 subscribe() 的 ReplayAdapter 也成立，对显式继承 Transport 的类继承来的空桩方法也成立，这两种情况都什么也不注册（见 4.1）
_current_token	`CancellationToken	None`

A.2 传输层

Transport 协议

Runtime-checkable Protocol，四个方法，通过 NullTransport 兜底，都可选实现：

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 返回 {"action": "continue" | "stop" | "new_instruction", "message"?: str}。

SdkTransport

基于回调的适配器——SDK 使用方的首选：

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

未设置的回调会回落到 NullTransport 行为（放行，MCP Skills 闸门确认除外，一律拒绝 / 空串 / 停止）。

build_compat_transport

从 8 个老式分立回调参数（confirmation_callback、step_callback、thinking_callback、ask_user_callback、output_callback、tool_complete_callback、llm_text_callback、on_max_iterations_callback）构造 Transport 的 helper。位于 agentao.embedding.compat，是仍在用 0.2.10 之前回调形态的宿主有文档保证的迁移面。自 0.5.0 起这些 kwarg 已从 Agentao(...) 签名中移除，所以它是继续使用这些回调的唯一途径：Agentao(transport=build_compat_transport(confirmation_callback=...), ...)。它本身没有被废弃。

AgentEvent / EventType

@dataclass
class AgentEvent:
    type: EventType
    data: dict

EventType 覆盖 UI 流式输出、工具生命周期、LLM 调用元数据、replay 可观测性和运行时状态变化。payload 完整参考见 4.2。

A.3 工具

Tool（抽象）

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   # 按工具选择启用，见 5.1

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

    # 注册时由 Agentao 注入
    working_directory: Path | None
    output_callback: Callable[[str], None] | None
    filesystem: FileSystem | None
    shell: ShellExecutor | None

    def _get_fs(self) -> FileSystem: ...      # 没注入时 lazy 构造 LocalFileSystem
    def _get_shell(self) -> ShellExecutor: ... # 没注入时 lazy 构造 LocalShellExecutor

自定义 Tool 子类需要读写文件时，调用 self._get_fs() 而不是直接用 pathlib——这样宿主注入的 FileSystem（Docker、虚拟 FS、审计代理）会被正确尊重。

ToolRegistry

方法	作用
register(tool: Tool, *, replace: bool = False, origin: str = "host")	注册。replace=False 冲突会 warn 后覆盖；replace=True 静默覆盖（打 INFO 审计）。底层 API——不绑定能力；优先用 Agentao(extra_tools=) / add_tool（见 5.1）。origin 记录工具来源（builtin / host / mcp / agent / plan），宿主保持默认 host；替换时覆盖
origin(name: str) -> str	读回记录的来源。子代理据此判定：host 工具只有在自身声明了 copies_to_subagents 时才会进入子代理，且以副本形式——这也包括用内置工具自己的类构造、替换了内置工具的宿主工具，正因如此判定不能看类身份。不经 register 放进 tools 的工具读作 host。名字不存在时抛 KeyError
unregister(name: str) -> bool	移除 name；返回是否存在过。纯 dict 操作
get(name: str) -> Tool	不存在时抛 KeyError，带可用工具列表
list_tools() -> list[Tool]
to_openai_format() -> list[dict]	OpenAI function-calling schemas

Capabilities（agentao.capabilities）

文件 / 搜索 / Shell 工具的 IO 都路由经过这两个 runtime-checkable Protocol。宿主通过 Agentao(filesystem=..., shell=...) 注入自定义实现，可以重定向到 Docker exec、虚拟文件系统、审计代理或远程 runner。包还导出与 0.2.16 之前字节级一致的默认实现 LocalFileSystem / LocalShellExecutor。

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]: ...      # 流式读取
    def write_text(self, path: Path, data: str, *, append: bool = False) -> None: ...
                                                               # 替换已有内容必须原子
    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: ...

冻结的 dataclass：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（默认 0）记录某个流没有保留的字节数，stdout_omitted_at / stderr_omitted_at（默认 0）给出它们在 stdout / stderr 中的字节偏移——0 表示在最前面，所以只保留尾部的执行器只需设置字节数。LocalShellExecutor 每个流保留开头和末尾各 512 KiB；完整保留输出的自定义执行器让这四个字段都保持 0。ShellRequest.cancellation_token（默认 None）是这一轮的 CancellationToken：它触发时，LocalShellExecutor 终止整个进程树，并返回 ShellResult.cancelled=True（默认 False）；忽略它的执行器照旧把命令运行到结束或超时。

ShellRequest 带的是一个 launch，不是一条命令（0.4.22 改动；以前 command、cwd、env 直接挂在它上面）。request.launch 是两种形状之一，执行器用 isinstance 分开：

形状	字段	怎么跑
LegacyLaunch	command、cwd、env、executable	subprocess.Popen(command, shell=True, executable=executable or None, cwd=cwd, env=env) —— 这就是过去每个宿主拿到的东西，没配置解释器时仍然是它
WindowsLaunch	application_name、command_line、cwd、env	用 command_line 启动 application_name，中间不经过 shell；shell 块点名解释器时产出这种

request.command 与 request.cwd 作为只读属性保留，所以只是记录 / 审计它们的执行器无需改动；request.env 没有了 —— 去 launch 上读。执行器还可以额外实现 ShellSpecProvider，即一个回答 ShellSpec | Exhausted 的 shell_spec 属性，来声明它将运行哪个解释器；命令地板扫描用的语法和提示词里的 shell 指引都读这个答案。完整契约见 docs/reference/host-api.zh.md。

宿主无法支持真正的后台执行时，可以在 run_background 抛 NotImplementedError——ShellTool 会把它呈现为普通的工具错误字符串。

write_text 除签名之外还带一条契约义务：替换已有文件内容的实现必须原子地完成，这样写到一半被打断（Ctrl+C、OOM kill）也绝不会把用户的文件留成截断状态。append=True 和目标尚不存在这两种情况仍走直写路径——它们都不可能破坏已有内容。LocalFileSystem 是参考实现：它在目标所在目录里暂存一个同级的 .{name}.*.tmp（审计 / watcher 包装层能观察到），再 os.replace 换入；对只读目标它会抛 PermissionError，而不是让目录级的 rename 权限悄悄绕过 chmod 444。

A.4 权限

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

0.2.16 起两个参数都显式传——project_root=None（或不传）会抛 TypeError。用户规则从 <user_root>/permissions.json 加载（一般 user_root=Path.home() / ".agentao"，对应 ~/.agentao/permissions.json）；传 user_root=None 可完全禁用用户态规则。project_root 仅用于检测（并 warn）一份残留的 <project_root>/.agentao/permissions.json —— 该文件不会作为规则源加载（原因见 5.4）。构造后用 set_mode() 切换预设模式。

方法	签名	作用
decide	`decide(tool_name: str, tool_args: dict) -> PermissionDecision	None`
set_mode	set_mode(mode: PermissionMode) -> None	切换预设（READ_ONLY、WORKSPACE_WRITE、FULL_ACCESS、PLAN）
active_mode	属性（只读）	当前生效的 PermissionMode

继承并 override decide() 可对接企业 IAM——确信度门控范例见 7.3。

A.5 记忆

MemoryManager（0.3.0 / #16 起）

MemoryManager(
    project_store: MemoryStore,                   # 必传，预先构造好的 store
    user_store: MemoryStore | None = None,        # 跨项目；None 表示禁用用户作用域
    guard: MemoryGuard | None = None,
)

构造器接收预先构造好的 MemoryStore 实例（嵌入式工厂负责构造并传入）。基于路径的构造下沉到 store 层：

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 —— 必传的项目作用域 MemoryStore（工厂用 SQLiteMemoryStore.open_or_memory(...)，目录写不进去会降级到 :memory: 而不是炸掉）
• user_store —— 可选的跨项目 store（工厂用 SQLiteMemoryStore.open(...)；传 None 完全禁用用户作用域，user-scope 写入会下移到 project）
• guard —— 可选 MemoryGuard，落盘前过滤敏感内容

方法	作用
upsert(req)	新增/更新一条记忆
save_from_tool(...)	save_memory 工具内部调用
get_entry(id) / get_all_entries(scope=?)	读
search(query, scope=?) / filter_by_tag(tag, scope=?)	搜索
delete(id) / delete_by_title(title)	软删除
clear(scope=?)	整个作用域软删除
save_session_summary(...) / get_recent_session_summaries(...)	压缩流水线用
archive_session(session_id=None) / clear_session()	会话末尾清理。给 archive_session 传会话自己的 id——摘要按它写入，跨会话尾部也只排除它，所以恢复的会话必须沿用自己的 id；不传参数得到一个新的随机 id
clear_all_session_summaries()	清掉所有会话的所有摘要
wipe_all()	硬清除：两个作用域的记忆（软删除，行仍留在库文件里）+ 所有会话摘要。永不抛异常，返回 MemoryWipeResult，成功信号是 ok。不清 review queue
get_stable_entries(...)	注入 <memory-stable> 系统提示块

MemoryStore（Protocol，0.3.0 / #16）

class MemoryStore(Protocol):
    # 记忆 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: ...
    # 会话摘要
    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: ...
    # 复审队列
    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 个方法，schema-less。实现这套 Protocol 即可把记忆后端换成 Redis / Postgres / 进程内 dict / 远端 API。默认 SQLiteMemoryStore 与 #16 之前的实现字节级一致。

可选、不在 Protocol 内：change_version() -> int，store 中的记忆被任何写入方（另一个 manager、另一个进程）写入时都会递增的计数器。MemoryRetriever 用它在这类写入后刷新召回索引。没有这个方法（或返回值不是 int）的 store 照常可用，只是召回只能看到经自己的 MemoryManager 写入的内容。SQLiteMemoryStore 用数据库文件里的触发器实现它。

SQLiteMemoryStore

SQLiteMemoryStore.open(path)              # 严格；磁盘出错抛
SQLiteMemoryStore.open_or_memory(path)    # 宽容；出错降级到 ":memory:"

两个 classmethod 取代了原本 MemoryManager.__init__ 的 try/except。项目 store 用 open_or_memory（宁可在内存里也别把 Agent 启动炸掉），用户 store 用 open（写不进去就该禁用，而不是默默把写入路由到项目作用域）。

A.6 MCP

MCPRegistry（Protocol，0.3.0 / #17）

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

MCP 服务器配置的来源。Agentao(mcp_registry=...) 是注入点；mcp_manager= 与 mcp_registry= 互斥（manager 是构造结果，registry 是配置源）。

具体实现	作用
FileBackedMCPRegistry(project_root, user_root=None)	CLI/ACP 默认。每次 list_servers() 都重读 <wd>/.agentao/mcp.json + <user_root>/mcp.json。同名冲突时用户胜；项目级条目仅可声明新 server name（冲突打 warning 并跳过项目项）。
InMemoryMCPRegistry(servers=None)	测试 / 嵌入式 host 用的程序化版本。构造入参做浅拷贝。

A.7 取消

CancellationToken

token = CancellationToken()
token.is_cancelled         # bool，不抛
token.cancel("reason")     # 幂等
token.check()              # 已取消则抛 AgentCancelledError
token.reason               # str

传给 agent.chat(..., cancellation_token=token)，就能从其他线程/异步任务中止。超时模式见 2.3。

AgentCancelledError

token 被取消时在 agent 循环里抛出。chat() 会捕获并返回 [Cancelled: reason] 字符串，而不是再往外抛。

A.8 ACP 客户端

ACPManager

宿主侧驱动 .agentao/acp.json 里声明的外部 ACP 服务器的典型入口。 如果你在无人值守环境里使用它（CI / worker / cron / queue consumer），那它就是正文里说的 Headless Runtime：不是新协议，也不是新类，只是 ACPManager 的一种运行方式。

方法	签名	作用
from_project	@classmethod from_project(project_root=None, *, terminal_auth=False) -> ACPManager	读 .agentao/acp.json。terminal_auth=True 声明 ACP Terminal Auth——仅当宿主能运行终端登录时传
add_server(name, config)		向运行中的 manager 登记新 server，保持停止；不影响运行中的 server
auth_methods(name)	-> list[dict]	该 server 最近一次 initialize 返回的 authMethods
needs_login(name)	-> bool	该 server 最近一次建会话回了 auth_required；重启后仍为真，直到成功打开会话或建会话因其他原因失败
reserve_for_login(name)	上下文管理器	登录 → 重启 → 连接期间，阻止其他线程在 name 上开 turn / 连接（SERVER_BUSY）
server_names	-> list[str]	已声明的服务器
start_all	start_all(only_auto=True)	启动所有 auto-start 服务器
start_server(name) / stop_server(name) / restart_server(name)
ensure_connected(name, cwd=?, mcp_servers=?)		幂等连接 + 新建会话
send_prompt(name, text, timeout=?, interactive=True, cwd=?, mcp_servers=?, interaction_policy=?)	-> dict（session/prompt 的结果）	在长连会话上跑一轮。fail-fast：已有活跃 turn 时抛 SERVER_BUSY（public）
prompt_once(name, prompt, cwd=?, mcp_servers=?, timeout=?, interactive=False, stop_process=True)	-> PromptResult	Fail-fast 一次性，自动清理（public）
send_prompt_nonblocking / finish_prompt_nonblocking / cancel_prompt_nonblocking		底层异步版本（internal / unstable，不在 embedding contract 内）
stop_all()		停所有子进程
get_status()	-> list[ServerStatus]	类型化 headless 快照（Week 1 核心字段冻结 + Week 2 增量字段）
readiness(name)	-> Literal["ready","busy","failed","not_ready"]	基于 state × active-turn 的类型化可用性分级（Week 2）
is_ready(name)	-> bool	readiness(name) == "ready" 的快捷方式
reset_last_error(name)		清除 manager 上记录的 last_error / last_error_at
get_client(name) / get_handle(name)		自省
config（属性）	-> 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

ACPManager.get_status() 的类型化返回值。Week 1 核心字段已冻结； Week 2 按增量方式扩展了诊断字段。

@dataclass(frozen=True)
class ServerStatus:
    # Week 1 —— 核心（冻结）
    server: str
    state: str
    pid: Optional[int]
    has_active_turn: bool

    # Week 2 —— 诊断（增量）
    active_session_id: Optional[str] = None
    last_error: Optional[str] = None
    last_error_at: Optional[datetime] = None   # 带 tzinfo=UTC
    inbox_pending: int = 0
    interaction_pending: int = 0
    config_warnings: List[str] = field(default_factory=list)

Week 2 字段语义要点：

• last_error 在成功 turn 之间保持，不会被覆盖成 None；如需清空， 调用 reset_last_error(name)。
• last_error_at 的时间戳由 manager 存入时取的 datetime.now(timezone.utc)，不是抛出时的瞬时。消费者应据此判断错误 是否陈旧，而不是把它当作精确 raise 时刻来使用。
• SERVER_BUSY 与 SERVER_NOT_FOUND 不写入 last_error，因为它们是 调用方侧信号，不是服务端状态。

各字段的确切含义和 readiness 分级见 3.4 状态字段 与就绪分级；state-vs-error 合约见 附录 D.5。

异常类

类	code	说明
AcpClientError	任意 AcpErrorCode	基类
AcpServerNotFound	SERVER_NOT_FOUND	同时继承 KeyError
AcpRpcError	.rpc_code: int + .error_code: PROTOCOL_ERROR	JSON-RPC 响应错
AcpInteractionRequiredError	INTERACTION_REQUIRED	带 prompt + options

完整错误码表见 附录 D。

load_acp_client_config

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

project_root 为 None 时退回 Path.cwd()。

不构造 manager 直接校验 + 加载 .agentao/acp.json。写 config lint 工具时好用。

A.9 技能

SkillManager

通过 from agentao import SkillManager 懒加载。一般通过 agent.skill_manager 使用。

方法	作用
list_available_skills() -> list[str]	当前可发现的技能名
list_all_skills() -> list[str]	含已禁用的
`get_skill_info(name) -> dict	None`
activate_skill(name, task_description)	激活——把 SKILL.md + 活跃的 reference 注入系统提示
enable_skill(name) / disable_skill(name)	持久化的启用/禁用

A.10 嵌入 Host 合约

自 0.3.1 起 Stable。包名于 0.4.2 由 agentao.harness 重命名为 agentao.host。

agentao.host 是嵌入 Agentao 的 稳定宿主面 API。内部运行时类型（AgentEvent、ToolExecutionResult、PermissionEngine）刻意 不在 该面内 —— 任何版本都可能改动。只针对 agentao.host（加上 Agentao(...) 构造器以及上文标注的方法）开发的宿主可在版本升级中保持前向兼容。

命名说明。 "Harness" 仍然指 Agentao 自身嵌入在宿主应用中运行 这一概念（见 docs/design/embedded-host-contract.md 设计语境）；合约包重命名为 agentao.host 是为了让 from agentao.host import HostEvent 读起来自洽。旧的 agentao.harness 导入路径以及旧符号名（HarnessEvent、HarnessReplaySink、export_harness_*）自 0.4.2 起告警，已于 0.5.0 移除 —— 把 harness / Harness 换成 host / Host 即可。

完整参考：docs/reference/host-api.md · docs/reference/host-api.zh.md。设计动机：docs/design/embedded-host-contract.md。

公共导出

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

符号	作用
ActivePermissions	当前权限策略的只读快照（mode、rules、loaded_sources）
ToolLifecycleEvent	单次工具调用的公共生命周期信封。phase ∈ {started, completed, failed}；取消以 phase="failed", outcome="cancelled" 体现
SubagentLifecycleEvent	子 Agent 任务/会话的血缘事实。phase ∈ {spawned, completed, failed, cancelled} —— cancelled 在这里是独立 phase。failed 有两义：子 Agent 抛了异常（error_type 是异常类名），或它跑完却没给出答案（error_type = "incomplete:<reason>"，<reason> ∈ no_output、reasoning_only、length_truncated、doom_loop、llm_error、max_iterations）。升级告警前先按 error_type 分支——见 4.7
SubagentUsage	终态 SubagentLifecycleEvent.usage 上的 token 计数（0.5.2）：prompt_tokens（整个输入）、completion_tokens、cache_read_tokens、cache_creation_tokens —— 后两项是 prompt_tokens 的组成部分 —— 以及自 0.5.15 起的 reasoning_tokens，它是 completion_tokens 的组成部分（线路不提供时为 0）。在 spawned 上、以及子 Agent 根本没建出来时为 None。事件触发时已计入父级的会话累计
PermissionDecisionEvent	单次权限决定的投影。在 allow / deny / prompt 都触发；不渲染 allow 的消费者也必须排空迭代器以避免背压
HostEvent	三种事件模型的判别联合（Pydantic discriminator: event_type）
RFC3339UTCString	受约束的时间戳类型。仅允许标准 Z 后缀 —— +00:00 偏移会被拒绝
EventStream	Agentao.events() 的运行时侧。仅供参考：Agentao 把自己的流设为私有，宿主通过 agent.events() 或 agent.add_host_event_observer() 读取，不自己调 publish() / subscribe()
StreamSubscribeError	同一 session_id 过滤器上发起第二个并发订阅时抛出，且第一个还开着；在第二个迭代器第一次迭代时抛出，而不是调用 agent.events() 时（MVP 每个过滤条件只支持一个消费者；先用 aclosing 关掉第一个，或者用同步观察者做多路分发）
export_host_event_json_schema()	导出事件 + 权限面的标准 JSON schema。tests/test_host_schema.py 用它与 docs/schema/host.events.v2.json 做字节相等校验
export_host_acp_json_schema()	导出宿主面 ACP 载荷的标准 JSON schema。快照在 docs/schema/host.acp.v1.json
CancellationToken	chat() / arun() 的 cancellation_token= 参数接受的 token。重新导出 agentao.cancellation.CancellationToken（同一个类）
Tool、AsyncToolBase	宿主自带工具的基类（extra_tools= / add_tool）。重新导出 agentao.tools.base 中的正式类型
RegistrableTool	Union[Tool, AsyncToolBase] —— extra_tools= / add_tool 接受的类型
TextDelta	Agentao.astream() 产出的一段 assistant 文本（text: str）。用于显示：增量拼起来不是回答。不是 HostEvent 成员；不投影进 replay，也不在 schema 快照里
TurnOutcome	一轮如何结束：text、status、incomplete_reason、tool_count、error、finish_reason_missing 以及 .is_answer。astream() 的最后一项，也是 agent.last_turn 的返回值。与 agentao.TurnOutcome 是同一个类

agent.events(session_id=None)

返回 HostEvent 的异步迭代器。传 session_id= 过滤；传 None 订阅该 Agentao 实例下所有会话。

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

投递契约：

• 同会话顺序保证；跨会话全局顺序不保证。
• 同一 tool_call_id 内，PermissionDecisionEvent 一定先于 ToolLifecycleEvent(phase="started")。
• 不 replay。 第一次订阅前发出的事件被丢弃。
• 背压走宿主拉取：消费者慢时，生产者会 await 匹配事件的容量，不会丢事件，也不会无限堆队列。
• 取消迭代器会释放队列/订阅资源。
• MVP 每个 session_id 过滤条件只支持一个异步迭代器消费者；第一个还开着时，同一过滤条件的第二个会在第一次迭代时抛出 StreamSubscribeError。需要多个消费者时，用 agent.add_host_event_observer()。

agent.active_permissions()

返回 JSON-safe 的 ActivePermissions 快照：

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

loaded_sources 是稳定的字符串标签：preset:<mode>、user:<path>、injected:<name>。（project:<path> 已不再发出 —— 项目级权限不会被加载，见 5.4。）MVP 不 暴露逐规则 provenance —— 需要细到规则级的宿主请把 loaded_sources 与自己注入的策略元数据组合。

未配置 permission_engine 时，运行时返回宽松回退：mode="workspace-write"、空 rules、loaded_sources=["default:no-engine"]。该标签明确告诉宿主："看到的是无引擎回退而非配置策略"：不执行任何规则，连 workspace-write 预设的规则也不执行，所有"问"都交给 transport（NullTransport 会一律批准）。

叠加策略的宿主调 agent.permission_engine.add_loaded_source("injected:<name>") 让快照反映其 provenance。set_mode() 与 add_loaded_source() 都会让缓存失效。

Schema 快照

每个发布版都附带 check-in 的 JSON schema 快照：

• docs/schema/host.events.v2.json —— 事件 + 权限面
• docs/schema/host.acp.v1.json —— 宿主面 ACP 载荷

tests/test_host_schema.py 会从 Pydantic 模型重新生成 schema，并与快照做字节相等比对。任何改变 wire form 的 model 变更必须在同一 PR 内同时更新模型与快照。新增 optional 字段向后兼容；删除/重命名需要 schema 版本号 bump。

不在合约内（明确不做）

• 公共 agent graph / descendants 存储 API
• 宿主面 hooks list/disable API（规则作者视角见 §5.7 插件 Hooks）
• 宿主面 MCP reload / 生命周期事件
• 本地 plugin export/import；远程 plugin 共享
• 外部会话 import
• 生成式客户端 SDK

CLI 可基于同一组事件构建自己的 UI，但其本地存储与命令不会被提升为 host 合约。

---

→ 附录 F · FAQ