AIP — Agent Infrastructure Protocol 分析

对 AIP v1 设计、架构与实现的深度解析 — 位于 MCP 等工具协议之上的治理控制面，为每一次 AI Agent 调用提供风险分级的能力管理、策略评估、配额执行与不可篡改的证据审计。

1. 为什么需要 AIP

现代 AI Agent 通过 MCP（Model Context Protocol）等协议调用外部工具。MCP 赋予 Agent 触达能力 — 读文件、调 API、执行命令。但它不回答以下问题：

• 谁被允许 调用这个工具？
• 在什么条件下 应修改或阻止调用？
• 调用之后发生了什么，能否证明？

AIP 填补这一空白。它不是 MCP 的替代品 — 而是其上层的 治理层。

代码证据：

• aip-spec/reference/aip_runtime/mcp_adapter.py — MCP 服务器作为执行后端被消费，而非被替换
• aip-spec/spec/aip-1.md — §1.1：“AIP 位于 Agent 运行时与执行适配器之间”

2. 设计原则

AIP 建立在五条基础原则之上，每条均可追溯到代码。

2.1 能力优先（Capability-First）

核心抽象是 Capability — 受治理的、有类型的、风险分级的工作单元。Agent 可调用的每个工具、API、函数都被包装为带有显式元数据的 Capability。

代码证据：

• aip-spec/reference/aip_runtime/models.py — @dataclass(frozen=True) class Capability，包含 capability_id、version、risk_tier、input_schema、output_schema、execution、lifecycle

2.2 治理原生（Governance-Native）

策略评估、审批工作流、配额执行、审计记录都是协议级原语 — 不是可选附加功能。

代码证据：

• aip-spec/reference/aip_runtime/runtime.py — invoke() 执行 5 个强制阶段；策略评估不可跳过
• aip-spec/reference/aip_runtime/policy_engine.py — _DEFAULT_BY_RISK：即使没有显式策略，CRITICAL 级默认拒绝

2.3 执行无关（Execution-Agnostic）

AIP 定义 语义，而非线格式。同一个 Capability 合约可分派到 MCP、REST、CLI、内部函数、容器或 Worker 队列。

代码证据：

• aip-spec/reference/aip_runtime/models.py — ExecutionBinding.type: str # mcp | rest | cli | internal | container | worker
• aip-spec/reference/aip_runtime/runtime.py — self._handlers: dict[str, ExecutionHandler] — 按传输类型可插拔

2.4 不可变设计（Immutability by Design）

所有核心数据模型均为冻结数据类。证据记录创建后不可修改。能力版本仅可追加。

代码证据：

• aip-spec/reference/aip_runtime/models.py — 每个模型均使用 @dataclass(frozen=True)
• aip-spec/reference/aip_runtime/evidence_store.py — record() 若 evidence_id 已存在则抛出 ValueError

2.5 故障关闭默认（Fail-Closed Defaults）

当没有策略显式允许某次调用时，风险等级决定默认行为。CRITICAL 操作默认拒绝。

代码证据：

• aip-spec/reference/aip_runtime/policy_engine.py：

_DEFAULT_BY_RISK = {
    "LOW": "allow",
    "MEDIUM": "allow",
    "HIGH": "require_approval",
    "CRITICAL": "deny",
}

3. 五阶段控制流

每次 AIP 调用都遵循确定性的五阶段管线。不可跳过任何阶段，顺序固定。

代码证据：

• aip-spec/reference/aip_runtime/runtime.py — invoke() 方法，第 64–195 行

阶段 1：解析能力（Resolve Capability）

注册表按请求的 capability_id 进行版本解析：

• 如指定版本：精确匹配
• 否则：最新的非废弃版本（active > staged > shadow）

代码证据：

• aip-spec/reference/aip_runtime/registry.py — get() 方法，状态优先级：{"active": 0, "staged": 1, "shadow": 2}

candidates.sort(
    key=lambda c: (
        status_priority.get(c.lifecycle.status, 99),
        -c.major_version,
        -c.minor_version,
    )
)

若未找到能力，控制流立即终止并返回 CAPABILITY_NOT_FOUND。

阶段 2：评估策略（Evaluate Policies）

所有匹配的策略按优先级顺序评估（数值越小优先级越高）。每条策略有目标规则和条件评估。

代码证据：

• aip-spec/reference/aip_runtime/policy_engine.py — evaluate() 方法：

# 终止性：deny
if rule.decision == "deny":
    return "deny", summaries, evaluations, {}

# 终止性：require_approval
if rule.decision == "require_approval":
    return "require_approval", summaries, evaluations, {}

# 累积性：modify
if rule.decision == "modify" and rule.modifications:
    # ... 合并修改

策略目标匹配

策略通过四个维度匹配调用：

代码证据：

• aip-spec/reference/aip_runtime/policy_engine.py — _policy_matches_target() 使用 fnmatch.fnmatch() 进行 glob 匹配

条件语言

叶节点条件使用 {field, operator, value} 结构，通过点路径解析调用请求中的字段。

支持的运算符：eq、ne、gt、lt、gte、lte、in、not_in、matches、starts_with、contains

组合条件：all_of（AND）、any_of（OR）、not_（NOT）— 递归求值。

代码证据：

• aip-spec/reference/aip_runtime/policy_engine.py — _evaluate_condition() 递归 AND/OR/NOT，_resolve_field() 点路径访问，_compare() 11 种运算符

风险等级单调性

关键安全不变量：策略只能使有效风险等级 更严格，不能降低。

# 不可低于能力声明的等级
if RISK_TIER_ORDER.get(candidate, 0) >= RISK_TIER_ORDER.get(
    capability.risk_tier, 0
):
    effective_risk_tier = most_restrictive_tier(
        effective_risk_tier, candidate
    )

阶段 3：检查配额（Check Quota）

跨多个维度的预算执行：

配额作用域：capability、actor、session、tenant 或 global。多层级配额并存时，最严格的限制生效。

代码证据：

• aip-spec/reference/aip_runtime/models.py — QuotaSpec 数据类，包含所有维度

阶段 4：分派执行（Dispatch Execution）

运行时将请求路由到对应传输类型的已注册处理器，并应用累积的策略修改。

transport_type = capability.execution.type if capability.execution else "internal"
handler = self._handlers.get(transport_type)
output = handler(capability, effective_input, effective_options)

若该传输类型没有注册处理器，返回 TRANSPORT_ERROR。

代码证据：

• aip-spec/reference/aip_runtime/runtime.py — 分派段，处理器查找与执行

阶段 5：记录证据（Record Evidence）

每次调用 — 无论成功或失败 — 恰好产生一条不可变的证据记录。

代码证据：

• aip-spec/reference/aip_runtime/evidence_store.py：

class EvidenceStore:
    """AIP-1 §3.4.1 保证：
    - 不可变：记录创建后不可修改。
    - 完整性：每次调用恰好产生一条记录。
    - 可审计：足够回答 谁/什么/何时/为何/成本。
    """

    def record(self, evidence: Evidence) -> None:
        if evidence.evidence_id in self._records:
            raise ValueError(
                f"Evidence {evidence.evidence_id} already exists (immutability violated)"
            )

证据记录包含：输入/输出的 SHA-256 摘要、完整的策略评估链、成本指标、执行详情、错误信息和参与者委托链。

4. 四大核心对象

AIP 精确定义四个核心领域对象，全部映射到 JSON Schema。

4.1 能力（Capability）

Agent 功能的基本单元 — 受治理的、有类型的、风险分级的操作。

代码证据：

• aip-spec/reference/aip_runtime/models.py — Capability 数据类，第 43–64 行
• aip-spec/spec/schemas/capability.json — 规范 JSON Schema

4.2 调用（Invocation — Request / Response）

事务单元：一个请求进，一个响应出，一条证据记录生成。

请求字段： invocation_id、capability_id、actor（含 delegation_chain）、input、context（run_id、plan_id、step_index）、options（dry_run、timeout、priority、trace）

响应字段： status（success | error | denied | timeout | circuit_open | quota_exceeded）、output、error、cost、evidence_id、policy_decisions

代码证据：

• aip-spec/reference/aip_runtime/models.py — InvocationRequest（第 98–105 行）、InvocationResponse（第 134–141 行）

4.3 策略（Policy）

声明式规则，治理能力调用。

@dataclass(frozen=True)
class Policy:
    policy_id: str
    target: PolicyTarget         # capabilities, risk_tiers, actors, actor_types
    rules: tuple[PolicyRule, ...]  # 有序，每个策略首条匹配
    description: str = ""
    enabled: bool = True
    priority: int = 100          # 越小越先评估

代码证据：

• aip-spec/reference/aip_runtime/models.py — Policy、PolicyRule、PolicyTarget、PolicyCondition、PolicyModifications，第 147–198 行

4.4 证据（Evidence）

为每次调用产生的不可变审计记录。

代码证据：

• aip-spec/reference/aip_runtime/models.py — Evidence 数据类，第 221–237 行
• aip-spec/reference/aip_runtime/evidence_store.py — compute_digest() 使用 hashlib.sha256

5. MCP 桥接 — 零修改集成

最重要的适配器：任何现有 MCP 服务器 无需任何修改 即可成为 AIP 执行后端。

5.1 架构

AIP Runtime.invoke()
    ↓
MCPAdapter.handle(capability, input, options)
    ↓
_StdioTransport（每个 MCP 服务器一个）
    ↓ JSON-RPC 2.0 over stdio
MCP 服务器（未修改）

代码证据：

• aip-spec/reference/aip_runtime/mcp_adapter.py — 类层级：MCPAdapter → _StdioTransport → 子进程

5.2 自动发现

MCP 工具被自动发现并注册为 AIP 能力：

def discover_capabilities(self, provider_id: str, ...) -> list[Capability]:
    result = transport.request("tools/list", {})
    tools = result.get("tools", [])
    # 每个 MCP 工具 → 一个 AIP Capability：
    #   capability_id = f"{provider_id}.{name}"
    #   risk_tier = 从工具名称/描述推断
    #   execution = ExecutionBinding(type="mcp", endpoint=provider_id, method=name)

代码证据：

• aip-spec/reference/aip_runtime/mcp_adapter.py — discover_capabilities()，第 296–346 行

5.3 风险等级推断

工具通过关键词分析自动分类：

代码证据：

• aip-spec/reference/aip_runtime/mcp_adapter.py — infer_risk_tier() 函数，第 231–238 行

5.4 服务器生命周期

每个 MCP 服务器作为子进程管理，具有完整的生命周期控制：

• 启动：子进程 + JSON-RPC 初始化握手（协议版本 2024-11-05）
• 健康检查：is_alive() 通过进程轮询
• 自动重启：最多 max_restarts 次（默认 3）
• 优雅停止：SIGTERM → 等待 5s → SIGKILL

代码证据：

• aip-spec/reference/aip_runtime/mcp_adapter.py — _StdioTransport 类，第 78–221 行

5.5 错误映射

MCP 错误被翻译为 AIP 错误码：

代码证据：

• aip-spec/reference/aip_runtime/mcp_adapter.py — _MCP_ERROR_MAP 字典，第 244 行

6. HTTP 绑定

AIP 暴露三个 HTTP 端点用于外部集成：

基于 FastAPI 构建，带 Pydantic 校验。刻意保持精简 — 注册管理、策略管理和配额管理端点推迟到后续版本。

代码证据：

• aip-spec/reference/aip_runtime/http_binding.py — create_app(runtime: AIPRuntime) -> FastAPI

@app.post("/invoke")
async def invoke(req: InvokeRequest) -> JSONResponse:
    """完整控制流：解析 → 策略 → 配额 → 执行 → 证据。"""
    aip_request = InvocationRequest(...)
    response = runtime.invoke(aip_request)
    status_code = 200 if response.status == "success" else 422
    return JSONResponse(content=result, status_code=status_code)

7. 错误分类

AIP 定义了包含 11 种错误码的结构化错误分类：

代码证据：

• aip-spec/reference/aip_runtime/models.py — InvocationError 数据类，含 code、message、retryable、retry_after_ms
• aip-spec/reference/aip_runtime/runtime.py — _fail() 方法构造错误响应

8. 与 OctopusOS 内核的关系

AIP 提取自 OctopusOS 治理基础设施。内核内部合约直接映射到 AIP 对象：

代码证据：

• kernel/contracts/policy_engine.py — RiskTaxonomy、BudgetQuotaSpec、PolicyDecision、ApprovalPipelineState
• kernel/governance/policy_engine.py — evaluate_policy_decision()
• kernel/governance/registry.py — validate_registry_snapshot()（故障关闭）
• kernel/governance/quota.py — evaluate_quota()（故障关闭）
• kernel/ports/evidence/interfaces.py — EvidenceIndexPort、ReplayLedgerPort

内核使用相同的冻结数据类模式、相同的风险等级排序、相同的故障关闭默认 — AIP 是在 OctopusOS 内部经过实战检验的治理模式的 公开规范。

9. 参考实现架构

参考实现包含 6 个模块，核心运行时零外部依赖：

代码证据：

• aip-spec/reference/aip_runtime/ — 全部 7 个模块
• aip-spec/reference/tests/ — 34 个测试覆盖核心流程

依赖关系图

10. 四场景演练

站点上的 AIP 动画演示了四个代表性场景，展示协议在不同风险等级下的行为。

场景 1：读取文件（LOW 风险 — 允许）

请求：  fs.file.read → path=/workspace/readme.md
策略：  无显式策略匹配
默认：  LOW 风险 → 允许
结果：  ✅ 已执行，证据已记录

场景 2：写入文件并修改策略（HIGH 风险 — 修改后执行）

请求：  fs.file.write → path=/workspace/output.txt
策略 1：pol_sandbox → allow（路径在 /workspace 内）
策略 2：pol_force_dry_run → modify（注入 dry_run: true）
结果：  ✅ 以 dry_run=true 执行，证据已记录

场景 3：沙箱外写入（HIGH 风险 — 拒绝）

请求：  fs.file.write → path=/etc/passwd
策略：  pol_sandbox → deny（路径在 /workspace 外）
结果：  ❌ POLICY_DENIED，执行从未到达，证据已记录

场景 4：删除操作（CRITICAL 风险 — 拒绝）

请求：  fs.file.delete → path=/workspace/temp.log
策略：  pol_no_critical_agent → deny（阻止 Agent 的 CRITICAL 操作）
默认：  即使无策略，CRITICAL 风险 → deny
结果：  ❌ POLICY_DENIED，证据已记录

关键洞察：在场景 3 和 4 中，执行阶段 从未到达。策略引擎在任何副作用发生前终止了控制流。但证据记录仍然被创建 — 无论结果如何，审计轨迹都是完整的。

11. 设计模式总结

12. 架构全景图