附录 A · API 参考

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

权威 __all__：

• from agentao import ... → Agentao、SkillManager
• from agentao.embedding import ... → build_from_environment
• from agentao.transport import ... → AgentEvent、EventType、Transport、NullTransport、SdkTransport、build_compat_transport
• 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
• from agentao.permissions import ... → PermissionEngine、PermissionMode、PermissionDecision
• from agentao.memory.manager import MemoryManager
• from agentao.cancellation import ... → CancellationToken、AgentCancelledError
• 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、PermissionDecisionEvent、RFC3339UTCString、export_host_event_json_schema、export_host_acp_json_schema —— 宿主面 harness 合约，详见 A.10

A.1 Agentao

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

构造器

完整参数表见 Part 2.2。0.3.0 起，不传 working_directory= 调用 Agentao() 会从 Python 签名分派直接抛 TypeError——软废弃周期已结束。完整嵌入式接入实践见 docs/guides/embedding.md。

Agentao(
    api_key: str | None = None,
    base_url: str | None = None,
    model: str | None = None,
    temperature: float | None = None,
    transport: Transport | None = None,
    *,
    working_directory: Path,                     # 0.3.0 起必传
    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,
    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,
    # 老式回调（已废弃 —— 会发 DeprecationWarning；0.5.0 移除）
    output_callback: Callable[[str], None] | None = None,
    confirmation_callback: Callable[[str, str, dict], bool] | None = None,
    ask_user_callback: Callable[[str], str] | None = None,
    on_max_iterations_callback: Callable[[int, list], dict] | None = None,
    # ...
)

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

• llm_client= 与任何 api_key= / base_url= / model= / temperature= 同时传
• mcp_manager= 与 extra_mcp_servers= 同时传
• mcp_manager= 与 mcp_registry= 同时传——registry 是配置源，manager 是构造结果
• enabled_tools= 与 disable_tools= 同时传——白名单和黑名单并存有歧义（enabled_tools=set() 也照样报错）。见 5.1

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

• replay_config=None —— 构造时不读 <wd>/.agentao/replay.json，内部用 no-op 的 ReplayConfig()。
• 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 子命令也会短路，并打印明确的提示。

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 变量、~/.agentao/permissions.json（项目级文件故意不加载 —— 见 5.4）、<wd>/.agentao/mcp.json + ~/.agentao/mcp.json（同名冲突时用户胜；项目级仅可新增）、内存目录，然后用发现到的值构造 Agentao。调用方传入的 **overrides 优先。同样会触发上面构造器的互斥校验（如果 overrides 含 llm_client，工厂会先把发现到的 api_key / base_url / model 丢掉再转发）。

方法

方法	签名	作用
chat	`chat(user_message: str, max_iterations: int = 100, cancellation_token: CancellationToken	None = None) -> str`
arun	`async arun(user_message: str, max_iterations: int = 100, cancellation_token: CancellationToken	None = None) -> str`
clear_history	clear_history() -> None	清 self.messages；不影响 memory DB
close	close() -> None	关 MCP 子进程与 DB handle；请放 finally:
set_provider	set_provider(api_key, base_url=None, model=None) -> None	运行时换 LLM
set_model	set_model(model: str) -> str	只换模型；返回旧 id
events (0.3.1+)	`events(session_id: str	None = None) -> AsyncIterator[HostEvent]`
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() 消失

属性

属性	类型	说明
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	Transport	当前传输层；可重新赋值
_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 行为（放行 / 空串 / 停止）。

build_compat_transport

从老式分立回调参数构造 Transport 的 helper。位于 agentao.embedding.compat，是仍在用 0.2.10 之前回调形态的宿主有文档保证的迁移面——显式调用它即可避开构造期的 DeprecationWarning。一般也不会直接用——Agentao(...) 在你传 confirmation_callback= 等参数时会自动调（并在每次构造时发一次 DeprecationWarning）。8 个 legacy 回调 kwarg 本身将在 0.5.0 从 Agentao(...) 签名中移除；build_compat_transport() 自身保留。

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

    @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)	注册。replace=False 冲突会 warn 后覆盖；replace=True 静默覆盖（打 INFO 审计）。底层 API——不绑定能力；优先用 Agentao(extra_tools=) / add_tool（见 5.1）
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(command, cwd, timeout, on_chunk, env)、ShellResult(returncode, stdout, stderr, timed_out)、BackgroundHandle(pid, pgid, command, cwd)。

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

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() / clear_session()	会话末尾清理
clear_all_session_summaries()	清掉所有会话的所有摘要
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 之前的实现字节级一致。

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) -> ACPManager	读 .agentao/acp.json
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, prompt, timeout=?)	-> PromptResult	交互式一轮（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，因为它们是 调用方侧信号，不是服务端状态。

完整 state-vs-error 合约和 readiness 分级见 docs/guides/headless-runtime.md。

异常类

类	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.5.0，首次导入时发一次 DeprecationWarning。

完整参考：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,
    PermissionDecisionEvent,
    RFC3339UTCString,
    export_host_event_json_schema,
    export_host_acp_json_schema,
)

符号	作用
ActivePermissions	当前权限策略的只读快照（mode、rules、loaded_sources）
ToolLifecycleEvent	单次工具调用的公共生命周期信封。phase ∈ {started, completed, failed}；取消以 phase="failed", outcome="cancelled" 体现
SubagentLifecycleEvent	子 Agent 任务/会话的血缘事实。phase ∈ {spawned, completed, failed, cancelled} —— cancelled 在这里是独立 phase
PermissionDecisionEvent	单次权限决定的投影。在 allow / deny / prompt 都触发；不渲染 allow 的消费者也必须排空迭代器以避免背压
HostEvent	三种事件模型的判别联合（Pydantic discriminator: event_type）
RFC3339UTCString	受约束的时间戳类型。仅允许标准 Z 后缀 —— +00:00 偏移会被拒绝
EventStream	Agentao.events() 的运行时侧。生产者调 publish()；消费者迭代 subscribe()
StreamSubscribeError	同一 session_id 过滤器上发起第二个并发订阅时抛出（MVP 每个 Agentao 只支持一个公共流消费者）
export_host_event_json_schema()	导出事件 + 权限面的标准 JSON schema。tests/test_host_schema.py 用它与 docs/schema/host.events.v1.json 做字节相等校验
export_host_acp_json_schema()	导出宿主面 ACP 载荷的标准 JSON schema。快照在 docs/schema/host.acp.v1.json

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 每个 Agentao 只支持一个公共流消费者。

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"]。该标签明确告诉宿主："看到的是无引擎回退而非配置策略"。

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

Schema 快照

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

• docs/schema/host.events.v1.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