4.1 Transport Protocol

本节你会学到
构成 Transport 接口的全部 4 个方法
每个方法的契约：哪个阻塞、哪个 fire-and-forget
NullTransport 的行为以及什么时候它就是合适的默认

Transport 是 Agent 运行时与宿主 UI/业务逻辑之间的唯一接口。理解它的四个方法，你就能在任何 UI 框架下集成 Agentao。

协议定义

python

@runtime_checkable
class CoreTransport(Protocol):
    # 单向事件（fire-and-forget）
    def emit(self, event: AgentEvent) -> None: ...

    # 阻塞式请求-响应
    def confirm_tool(self, tool_name: str, description: str, args: dict) -> bool: ...
    def ask_user(
        self,
        question: str,
        *,
        header: Optional[str] = None,
        options: Optional[List[str]] = None,
        multiple: bool = False,
        allow_custom: bool = True,
    ) -> str: ...
    def on_max_iterations(self, count: int, messages: list) -> dict: ...


@runtime_checkable
class Transport(CoreTransport, Protocol):
    # fan-out；省略它的 transport 只是 CoreTransport
    def subscribe(self, listener: Callable[[AgentEvent], None]) -> Callable[[], None]: ...

关键设计：

两者都是 Protocol（PEP 544）——你不必继承任何基类，实现这 4 个必填方法就能传给 Agentao(transport=...)
@runtime_checkable 让 isinstance(x, CoreTransport) 和 isinstance(x, Transport) 可用（但不保证方法类型正确；类型检查应依赖静态工具）
四个必填方法一分为二：1 个单向推送事件 + 3 个同步问答
subscribe(listener) 是可选的——Transport 里有声明，CoreTransport 里没有；NullTransport 和 SdkTransport 通过组合 EventBroadcaster helper 提供；自定义实现可以不实现。访问前用 getattr(transport, "subscribe", None) 探测——但方法存在不等于它真能用（见下一条）。
对静态类型检查器来说，四个必需方法单独构成一个协议 CoreTransport（agentao.transport），Transport 在它之上加了 subscribe。Agentao(transport=...) 的类型是 CoreTransport，所以没有 subscribe() 的 transport 在这里也能通过类型检查。运行时同理：对这样的 transport，isinstance(x, Transport) 为 False（runtime_checkable 检查要求每个成员都在，包括 subscribe），应改用 isinstance(x, CoreTransport)。反过来不成立：一个显式继承 Transport、却没有定义 subscribe() 的类能通过 isinstance(x, Transport)，但继承的是协议里的桩方法，它什么也不注册，返回 None——要可用的 subscribe()，继承 NullTransport（或组合 EventBroadcaster）。只有 Agentao.astream() 需要 subscribe()，缺了它会抛 TypeError。

方法一：`emit(event)` — 推事件

python

def emit(self, event: AgentEvent) -> None:
    """接收运行时事件。不得抛异常；错误必须吞掉。"""

契约：

Agent 在关键节点调用 emit（一轮开始、工具开始/输出/结束、LLM 流式文本、思考、错误…）
实现禁止抛异常——抛了会被上游 try/except 吞掉，但可能破坏状态一致性
实现应快速返回——这是同步调用，慢了会拖慢整个 Agent

典型实现：

python

def emit(self, event: AgentEvent) -> None:
    try:
        self._queue.put_nowait(event)   # 扔到队列让别的线程处理
    except Exception:
        pass  # 永不抛

事件类型全表见 4.2 AgentEvent 事件清单。

方法二：`confirm_tool(name, desc, args)` — 工具确认

python

def confirm_tool(self, tool_name: str, description: str, args: dict) -> bool:
    """询问是否允许这个工具执行。
    True  → 允许
    False → 取消（Agent 收到 "Tool execution declined" 字符串，继续推理）
    """

何时被调用：

Agent 准备调用某个 requires_confirmation=True 的工具时
默认触发者：write_file、run_shell_command、web_fetch、web_search

阻塞语义：这是同步调用——在你返回 True/False 前，Agent 的执行线程会停在这里。如果你的宿主是异步 UI，需要在 Transport 实现内部做阻塞等待（见 4.5）。

返回 False 的后果：

工具不执行
Agent 收到一个"用户取消"的假结果
LLM 会基于这个结果继续推理（通常会换个思路或停下来汇报）

方法三：`ask_user(question)` — 向用户反问

python

def ask_user(
    self,
    question: str,
    *,
    header: Optional[str] = None,
    options: Optional[List[str]] = None,
    multiple: bool = False,
    allow_custom: bool = True,
) -> str:
    """让 Agent 向用户反问一个开放问题，返回用户的回答。"""

question 始终是自由文本。仅限关键字的 header、options、multiple、allow_custom 是建议性提示，transport 可以渲染成选择题，也可以忽略。运行时，只接受 question 的方法照样能用：内置工具会检查 transport 方法本身的签名，只在它接受某个关键字参数时才转交对应的提示（transport/sdk.py::invoke_ask_user_callback），子代理提出的问题也一样。类型检查器拿一个类和 CoreTransport 比对时要求完整签名，所以新代码请声明这四个关键字参数（或写 **hints）。

何时被调用：

Agent 主动调用内置工具 ask_user 时
典型场景：信息不足、多选一决策、要求澄清歧义需求

默认兜底：NullTransport 返回固定字符串 "[ask_user: not available in non-interactive mode]"，Agent 收到后会据此决定继续/放弃。

方法四：`on_max_iterations(count, messages)` — 迭代上限兜底

python

def on_max_iterations(self, count: int, messages: list) -> dict:
    """Agent 达到 max_iterations（默认 100）时调用。
    返回 dict，key "action" 必填：
        "continue"        — 再给 N 轮继续跑
        "stop"            — 终止，返回当前结果
        "new_instruction" — 注入一条新 user 消息，需带 "message" 键
    """

经典用法：

python

def on_max_iterations(self, count, messages):
    # 自动续一次
    if not hasattr(self, "_continued"):
        self._continued = True
        return {"action": "continue"}
    # 续过了，还卡着 → 让 LLM 总结并停下
    return {
        "action": "new_instruction",
        "message": "请基于目前的信息给出最终答复，不要再调用工具。",
    }

详细策略见 4.6 最大迭代数兜底。

可选方法：`subscribe(listener)` —— 不重发地 fan-out

python

def subscribe(
    self, listener: Callable[[AgentEvent], None]
) -> Callable[[], None]:
    """注册一个额外 listener，每个被 emit 的事件都会回调它。
    返回一个幂等的 unsubscribe 函数。"""

为什么需要它 —— replay 录制器、审计流水线、指标采集等都需要观察所有事件，但不应取代主 Transport。subscribe() 让多个消费者挂到同一个 transport 实例上，省去手动 fan-out。

何时调用：

宿主在构造期挂一个 listener，直到调用返回的 unsubscribe 才解除注册
通知时以快照方式遍历 listeners，所以 emit 过程中订阅/取消订阅是安全的
listener 抛出的异常会被吞掉——绝不能污染运行时的 emit 路径

框架内置使用方：

replay 录制器（订阅 TURN_BEGIN / TURN_END / 工具 / 子 agent 事件，替代过去从 agent 状态直接调 replay adapter 的路径）
agent.events() 背后的宿主事件流（见 4.7）
agent.astream()：通过订阅（而不是替换 transport）流式输出一轮的 assistant 文本（见 4.7）。transport 没有 subscribe() 时，astream() 在调用时、本轮开始前就抛 TypeError。

调用前先探测 —— 路径 C 的自定义实现可能完全不实现该方法：

python

sub = getattr(transport, "subscribe", None)
if sub is not None:
    unsubscribe = sub(my_listener)
    try:
        ...
    finally:
        unsubscribe()

NullTransport 和 SdkTransport 通过组合 agentao.transport.EventBroadcaster 拿到 subscribe()。从零实现的 transport（如 ACP）想要 fan-out，可以照搬同样的写法：

python

from agentao.transport import EventBroadcaster

class MyTransport:
    def __init__(self):
        self._broadcast = EventBroadcaster()

    def emit(self, event):
        # ... 你真正的发送路径 ...
        self._broadcast.notify(event)

    def subscribe(self, listener):
        return self._broadcast.subscribe(listener)

三种实现路径

路径	适合场景	复杂度
用 `SdkTransport` + 回调	90% 嵌入场景	最低
继承 `NullTransport` 覆盖部分方法	只关心某几个事件	低
从零实现 `Transport` Protocol	完全自定义（如 ACP、消息队列桥接）	中

路径 A · SdkTransport

参见 4.3：

python

from agentao.transport import SdkTransport

transport = SdkTransport(
    on_event=handle,
    confirm_tool=approve,
    ask_user=prompt,
    on_max_iterations=bail_out,
)

路径 B · 继承 NullTransport

当你只关心部分事件且想显式控制每个方法时：

python

from agentao.transport import NullTransport, EventType

class MyTransport(NullTransport):
    def __init__(self, on_token):
        super().__init__()          # 初始化 subscribe()；astream() 依赖它
        self.on_token = on_token

    def emit(self, event):
        if event.type == EventType.LLM_TEXT:
            self.on_token(event.data["chunk"])
        super().emit(event)         # 仍然通知订阅者

    def confirm_tool(self, name, desc, args):
        # 只允许读类工具
        return name.startswith("read_") or name == "glob"

路径 C · 从零实现

最典型的真实例子：ACP 服务端。它不继承任何基类，而是把每次 emit 转成 session/update 通知、把 confirm_tool 转成 session/request_permission 请求发给 ACP Client。

python

class MyCustomTransport:
    """把 Agent 事件桥到你自己的消息协议。"""
    def __init__(self, send_to_client):
        self.send = send_to_client

    def emit(self, event):
        self.send({"type": "agent_event",
                   "event": event.type.value,
                   "data": event.data})

    def confirm_tool(self, name, desc, args):
        # 发请求给客户端，同步等响应
        return self.send({"type": "confirm", ...}, wait=True)

    def ask_user(self, q, **hints):
        return self.send({"type": "ask", "question": q, **hints}, wait=True)

    def on_max_iterations(self, count, msgs):
        return {"action": "stop"}

线程与异步注意事项

同步 Agent 线程里调用 Transport：所有 4 个方法在 Agent 的 chat() 循环里被同步调用
如果你的宿主是 asyncio：
- emit 可以 asyncio.run_coroutine_threadsafe(...) 回主循环
- confirm_tool / ask_user 需要跨线程同步等待（见 4.5 实现模式）

测试你的 Transport

python

from agentao.transport import AgentEvent, EventType

def test_my_transport():
    t = MyTransport()
    # 1. emit 不得抛
    t.emit(AgentEvent(EventType.LLM_TEXT, {"chunk": "hi"}))
    # 2. confirm_tool 必须返回 bool
    assert isinstance(t.confirm_tool("x", "", {}), bool)
    # 3. ask_user 必须返回 str
    assert isinstance(t.ask_user("q?"), str)
    # 4. on_max_iterations 必须返回带 action 的 dict
    r = t.on_max_iterations(100, [])
    assert r["action"] in {"continue", "stop", "new_instruction"}

TL;DR

Transport = 4 个方法：emit（fire-and-forget）·confirm_tool（阻塞 bool）·ask_user（阻塞 str）·on_max_iterations（阻塞 dict）。
emit 抛出的异常会被吞；其他三个的异常会向上传播。
NullTransport = 静默 + 自动批准——适合测试和无人值守批处理。MCP Skills 闸门发起的确认例外：它一律拒绝，因为这种同意只能由人给出（见 4.5）。
自己实现 Transport 时 4 个方法都要给（哪怕是 no-op stub），让 Agent 循环在所有路径上都安全。

→ 下一节：4.2 AgentEvent 事件清单

4.1 Transport Protocol ​

协议定义 ​

方法一：emit(event) — 推事件 ​

方法二：confirm_tool(name, desc, args) — 工具确认 ​

方法三：ask_user(question) — 向用户反问 ​

方法四：on_max_iterations(count, messages) — 迭代上限兜底 ​

可选方法：subscribe(listener) —— 不重发地 fan-out ​

三种实现路径 ​

路径 A · SdkTransport ​

路径 B · 继承 NullTransport ​

路径 C · 从零实现 ​

线程与异步注意事项 ​

测试你的 Transport ​

TL;DR ​

4.1 Transport Protocol

协议定义

方法一：`emit(event)` — 推事件

方法二：`confirm_tool(name, desc, args)` — 工具确认

方法三：`ask_user(question)` — 向用户反问

方法四：`on_max_iterations(count, messages)` — 迭代上限兜底

可选方法：`subscribe(listener)` —— 不重发地 fan-out

三种实现路径

路径 A · SdkTransport

路径 B · 继承 NullTransport

路径 C · 从零实现

线程与异步注意事项

测试你的 Transport

TL;DR