AI Agent开放协议实战：构建发现、协商与支付的完整闭环

各位读者朋友，大家好。最近 AI 领域的进展非常快，尤其是以大模型为核心的 Agent（智能体）应用，已经不再满足于“单打独斗”式地完成对话或工具调用。越来越多的业务场景中，我们能看到多个 AI Agent 协作完成任务：一个 Agent 负责拆解需求，另一个 Agent 负责调用搜索或数据库工具，还有一个 Agent 负责汇总结果。

但随之而来一个非常现实的问题：当 Agent 需要跨系统、跨团队、甚至跨公司协作时，它们如何找到彼此？如何就“帮什么忙、代价是什么”达成一致？又如何完成服务费用的支付？

近期在技术社区看到不少团队开始关注“Open protocol for AI agents to discover, negotiate, and pay each other”（用于 AI 智能体相互发现、协商和支付的开放协议）。本文将围绕这一方向，整理一套可供参考的闭环实现思路。我会从概念拆解、协议分层设计、环境准备、核心代码实现、支付对接、常见问题到最佳实践逐一展开，希望能给你提供一个相对完整的工程化视角。

1. 背景与核心概念

1.1 什么是 AI Agent 经济

AI Agent 经济（Agentic Economy）并不是一个遥远的概念。简单来说，当 AI Agent 具备自主决策能力后，它们之间的交互就不再局限于“API 调用”或“函数回调”，而是变成了一种带有业务属性、商务属性的协作关系。

设想一个场景：

• 一个“订餐 Agent”需要查询附近餐厅的实时排队情况。
• 它发现“餐厅数据服务 Agent”可以提供这类信息。
• 订餐 Agent 需要先完成注册、鉴权，然后发起需求描述。
• 餐厅数据服务 Agent 返回报价，双方协商计费方式。
• 订餐 Agent 确认价格后，通过协议完成支付。
• 餐厅数据服务 Agent 收到款项后，返回结构化数据。

这整个闭环，就是 AI Agent 经济的一个典型缩影。Open protocol 的核心目标，就是把“发现、协商、支付”这三个环节标准化，让不同团队开发的 Agent 可以像 Web 服务一样互相集成。

1.2 协议要解决的核心问题

过去，两个 Agent 协作通常采用“点对点定制开发”的方式。A 系统需要对接 B 系统时，必须知道 B 系统的接口地址、接口参数、认证方式、计费规则，甚至还需要单独约定对账方案。这种方式的缺点非常明显：

1. 无法动态发现：A 系统无法在运行时感知 B 系统是否在线、是否支持新的能力。
2. 协商成本高：价格、响应时间、服务级别（SLA）通常是写死的，无法根据请求上下文动态调整。
3. 支付链路断裂：Agent 无法自主完成“服务产生费用 -> 费用确认 -> 资金转移”的整条链路，往往需要人工介入。

开放协议通过标准化的消息结构、握手流程和结算接口，把上述问题抽象成了三个层面：

1.3 为什么需要掌握这套协议

从工程角度来看，理解这套协议的意义不只是“跟上热点”。它背后涉及的技术组件非常经典：身份认证、消息队列、状态机设计、幂等控制、分布式事务、回调通知等。无论你是后端开发、架构师，还是 AI 应用工程师，把这些能力落地到 Agent 场景中，都能迁移到其他分布式系统中。

本文的代码示例将围绕一个简化版的协议实现展开，重点演示“发现-协商-支付”的核心流程。考虑到不同团队的技术栈不同，我会以 Python 作为示例语言，同时把协议层的接口设计与业务逻辑拆开，方便你迁移到 Java、Go 或 Node.js。

2. 协议分层设计与核心术语

2.1 协议的分层结构

完整的 Agent 开放协议建议拆分为三层，避免所有逻辑耦合在一起。

• 传输层（Transport Layer）：负责 Agent 之间消息的可靠传递。本文示例中采用 HTTP + JSON 的方式，工程上也可以替换为 gRPC 或消息队列。
• 能力层（Capability Layer）：负责定义“Agent 能提供什么服务”。例如，一个天气 Agent 的 Capability 可以是weather.query，输入参数包含city和date。
• 商务层（Commerce Layer）：负责定义“服务怎么卖”。包括定价模式（按次、按时长、按数据量）、报价有效期、支付地址、退款策略等。

这样的分层带来一个好处：底层的传输协议如果更换，上层的能力描述和商务规则可以保持不变。

2.2 核心术语定义

在设计协议前，先统一几个术语：

举个例子：Consumer 发出一个 RFS，内容是“我需要北京市今天下午 3 点到 5 点的降雨概率，要求 JSON 格式”。Provider 收到后返回 Proposal，内容包括“可以提供该数据，单价 0.01 元/次，预计响应时间 200ms”。Consumer 确认后，生成 Agreement；Provider 完成查询后，发出 Invoice；Consumer 根据 Invoice 调用支付接口，生成 Receipt。

2.3 与普通 API Gateway 的区别

有读者可能会问：“这不就是 API 网关加一层计费吗？”其实有本质区别。

普通 API Gateway 的消费者和提供者事先都知道对方，接口文档是静态的。而 Agent 开放协议需要支持运行时动态发现和语义协商。也就是说，Consumer 在发起请求前，可能根本不知道 Provider 的接口长什么样，需要通过协议内的语义描述（例如 OpenAPI 片段或 JSON Schema）来动态生成调用参数。

3. 环境准备与基础架构

3.1 技术选型

本文的示例代码将采用以下环境：

• 操作系统：macOS 或 Linux 均可，Windows 建议使用 WSL2。
• 语言版本：Python 3.10+。
• Web 框架：FastAPI，用于构建 Provider 和 Consumer 的 HTTP 服务。
• 数据存储：SQLite，用于保存 Agent 注册信息、协议记录和账单。生产环境建议替换为 PostgreSQL。
• HTTP 客户端：httpx，用于 Agent 之间的异步通信。
• 支付模拟：Stripe 的测试模式，或者本地 Mock 支付服务。本文将以 Mock 支付服务演示完整闭环，真实接入时替换为合规支付渠道即可。

版本需要根据你的项目实际情况调整，本文示例以常见环境为例，重点演示配置思路。

3.2 项目结构

建议按模块分离设计，便于后续扩展：

agent-protocol-demo/ ├── protocol/ │ ├── models.py # 数据模型定义 │ ├── discovery.py # Agent 发现逻辑 │ ├── negotiation.py # 协商逻辑 │ ├── payment.py # 支付逻辑 │ └── messages.py # 标准消息体 ├── provider/ │ ├── main.py # Provider 服务入口 │ ├── capabilities.py # 能力注册与执行 │ └── config.py # Provider 配置 ├── consumer/ │ ├── main.py # Consumer 服务入口 │ └── client.py # Consumer 客户端逻辑 ├── payment_gateway/ │ └── mock_gateway.py # Mock 支付服务 ├── requirements.txt └── README.md

3.3 安装依赖

mkdir agent-protocol-demo && cd agent-protocol-demo python3 -m venv venv source venv/bin/activate pip install fastapi uvicorn httpx pydantic

这里的核心依赖是pydantic，它可以帮助我们实现协议消息体的校验，确保不同 Agent 之间传递的数据符合规范。

3.4 通用数据模型

协议的第一步，是定义统一的数据模型。创建protocol/models.py：

# 文件路径：protocol/models.py from enum import Enum from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class AgentStatus(str, Enum): AVAILABLE = "available" BUSY = "busy" OFFLINE = "offline" class PricingType(str, Enum): FIXED = "fixed" # 固定价格 PER_CALL = "per_call" # 按次计费 PER_TOKEN = "per_token" # 按 Token 计费 CUSTOM = "custom" # 自定义规则 class AgentCapability(BaseModel): """Agent 能力描述。""" name: str description: str = "" input_schema: Optional[Dict[str, Any]] = None output_schema: Optional[Dict[str, Any]] = None class AgentOffer(BaseModel): """服务提供方发布的服务目录。""" agent_id: str agent_name: str capabilities: List[AgentCapability] pricing_type: PricingType price: float = 0.0 currency: str = "USD" endpoint: str # 访问地址 status: AgentStatus = AgentStatus.AVAILABLE class ServiceRequest(BaseModel): """消费方发出的服务请求（RFS）。""" request_id: str = Field(default_factory=lambda: uuid4().hex) consumer_id: str capability: str parameters: Dict[str, Any] = {} max_budget: Optional[float] = None required_sla_ms: Optional[int] = None class Proposal(BaseModel): """提供方给出的报价提案。""" proposal_id: str = Field(default_factory=lambda: uuid4().hex) request_id: str provider_id: str capability: str price: float currency: str estimated_response_ms: int valid_until: str class Agreement(BaseModel): """双方达成的服务合约。""" agreement_id: str = Field(default_factory=lambda: uuid4().hex) request_id: str proposal_id: str provider_id: str consumer_id: str capability: str price: float currency: str status: str = "pending" # pending / active / completed / cancelled class Invoice(BaseModel): """服务完成后的账单。""" invoice_id: str = Field(default_factory=lambda: uuid4().hex) agreement_id: str provider_id: str consumer_id: str amount: float currency: str status: str = "unpaid" class Receipt(BaseModel): """支付完成后的凭证，用于对账。""" receipt_id: str = Field(default_factory=lambda: uuid4().hex) invoice_id: str payment_ref: str amount: float currency: str status: str = "paid"

这里要注意，uuid4().hex需要在文件顶部导入uuid：

from uuid import uuid4

数据模型定义完成后，接下来就可以围绕这些模型实现发现、协商和支付逻辑。

4. 完整实战：实现一个基于 Open Protocol 的 Agent 协作闭环

4.1 构建发现服务（Discovery）

发现服务的作用是维护一个 Agent 注册目录。每个 Provider 启动时，会向目录注册自己的能力和报价。

先创建protocol/discovery.py：

# 文件路径：protocol/discovery.py import json import sqlite3 from typing import List from .models import AgentOffer class DiscoveryServer: """基于 SQLite 的简易 Agent 注册中心。""" def __init__(self, db_path: str = "agent_registry.db"): self.conn = sqlite3.connect(db_path) self.conn.row_factory = sqlite3.Row self._init_table() def _init_table(self): self.conn.execute(""" CREATE TABLE IF NOT EXISTS agent_offers ( agent_id TEXT PRIMARY KEY, agent_name TEXT, offer_json TEXT, status TEXT ) """) self.conn.commit() def register(self, offer: AgentOffer): """注册或更新 Agent 能力。""" self.conn.execute( "INSERT OR REPLACE INTO agent_offers (agent_id, agent_name, offer_json, status) VALUES (?, ?, ?, ?)", (offer.agent_id, offer.agent_name, offer.model_dump_json(), offer.status.value) ) self.conn.commit() def discover(self, capability: str) -> List[AgentOffer]: """按能力名称查找可用 Agent。""" rows = self.conn.execute("SELECT offer_json FROM agent_offers WHERE status='available'").fetchall() result = [] for row in rows: offer = AgentOffer.model_validate_json(row["offer_json"]) caps = [c.name for c in offer.capabilities] if capability in caps: result.append(offer) return result

在 FastAPI 中注册对应的路由。创建provider/main.py时，我们先把 Provider 和 Discovery Server 放在同一个进程里，方便演示。

4.2 构建 Provider 服务

Provider 有两个职责：

1. 启动时注册自身能力。
2. 处理 Consumer 发来的协商请求、执行具体任务并出具账单。

创建provider/main.py：

# 文件路径：provider/main.py from fastapi import FastAPI from pydantic import BaseModel from protocol.models import ( AgentCapability, AgentOffer, Agreement, Invoice, Proposal, ServiceRequest, ) from protocol.discovery import DiscoveryServer app = FastAPI() discovery = DiscoveryServer() # 当前 Provider 的固定信息 PROVIDER_ID = "provider-weather-001" # 注册服务 @app.on_event("startup") def startup(): offer = AgentOffer( agent_id=PROVIDER_ID, agent_name="Weather Data Agent", capabilities=[ AgentCapability( name="weather.query", description="查询指定城市在指定日期的天气情况", input_schema={ "type": "object", "properties": { "city": {"type": "string"}, "date": {"type": "string"} }, "required": ["city"] } ) ], pricing_type="per_call", price=0.01, currency="USD", endpoint="http://localhost:9001" ) discovery.register(offer) # 接收消费方的服务请求（RFS） @app.post("/v1/negotiate") def negotiate(req: ServiceRequest): """ 处理 Consumer 发来的协商请求。 这里简化了协商策略，直接返回固定报价。 """ proposal = Proposal( request_id=req.request_id, provider_id=PROVIDER_ID, capability=req.capability, price=0.01, currency="USD", estimated_response_ms=80, valid_until="2030-12-31T23:59:59Z" ) return proposal # 执行服务并出具账单 @app.post("/v1/execute") def execute(agreement: Agreement): """ 根据 Agreement 执行实际任务。 现实中这里会调用大模型或内部服务。 """ if agreement.status != "active": return {"error": "agreement not active"} # 模拟业务逻辑 result = simulate_weather_query(agreement.capability) invoice = Invoice( agreement_id=agreement.agreement_id, provider_id=agreement.provider_id, consumer_id=agreement.consumer_id, amount=agreement.price, currency=agreement.currency ) return {"result": result, "invoice": invoice} def simulate_weather_query(capability: str): return { "city": "Beijing", "condition": "cloudy", "precipitation_probability": 0.25, "temperature_c": 18 }

4.3 构建 Consumer 客户端

Consumer 是整个流程的发起方。它需要完成以下步骤：

1. 调用发现服务，找到满足能力要求的 Provider。
2. 向 Provider 发出ServiceRequest。
3. 接收Proposal，判断价格是否在预算内。
4. 生成Agreement，并与 Provider 确认。
5. 接收Invoice。
6. 调用支付网关完成支付，获取Receipt。

创建consumer/client.py：

# 文件路径：consumer/client.py import httpx from protocol.models import ServiceRequest, Agreement, Receipt DISCOVERY_URL = "http://localhost:9000" PAYMENT_GATEWAY_URL = "http://localhost:9003" def create_client(): return httpx.Client(timeout=10.0) def discover_agents(client: httpx.Client, capability: str): resp = client.get(f"{DISCOVERY_URL}/discover", params={"capability": capability}) resp.raise_for_status() return resp.json() def send_rfs(client: httpx.Client, provider_endpoint: str, req: ServiceRequest): resp = client.post(f"{provider_endpoint}/v1/negotiate", json=req.model_dump()) resp.raise_for_status() return resp.json() def confirm_agreement(client: httpx.Client, provider_endpoint: str, agreement: Agreement): resp = client.post(f"{provider_endpoint}/v1/execute", json=agreement.model_dump()) resp.raise_for_status() return resp.json() def pay_invoice(client: httpx.Client, invoice: dict): payload = { "invoice_id": invoice["invoice_id"], "amount": invoice["amount"], "currency": invoice["currency"] } resp = client.post(f"{PAYMENT_GATEWAY_URL}/v1/charge", json=payload) resp.raise_for_status() return Receipt.model_validate(resp.json())

为了直观演示完整链路，创建一个run_consumer.py脚本：

# 文件路径：run_consumer.py import httpx from consumer.client import ( confirm_agreement, discover_agents, pay_invoice, send_rfs, ) from protocol.models import Agreement, ServiceRequest def main(): client = httpx.Client(timeout=10.0) # 1. 发现 agents = discover_agents(client, "weather.query") print("发现可用 Agent:", agents) if not agents: print("没有找到可用服务") return agent = agents[0] # 2. 协商 req = ServiceRequest( consumer_id="consumer-demo-001", capability="weather.query", parameters={"city": "Beijing"}, max_budget=0.05, required_sla_ms=500 ) provider_endpoint = agent["endpoint"] proposal = send_rfs(client, provider_endpoint, req) print("收到报价:", proposal) # 3. 判断预算 if proposal["price"] > req.max_budget: print("超出预算，协商取消") return # 4. 创建合约 agreement = Agreement( request_id=req.request_id, proposal_id=proposal["proposal_id"], provider_id=proposal["provider_id"], consumer_id=req.consumer_id, capability=req.capability, price=proposal["price"], currency=proposal["currency"], status="active" ) # 5. 执行并获取账单 exec_result = confirm_agreement(client, provider_endpoint, agreement) print("执行结果:", exec_result) invoice = exec_result["invoice"] # 6. 支付 receipt = pay_invoice(client, invoice) print("支付凭证:", receipt.model_dump()) client.close() if __name__ == "__main__": main()

4.4 构建 Mock 支付网关

支付环节是整个协议中最敏感的部分。真实业务中需要接入 Stripe、PayPal 或其他合规支付服务商。本文为了聚焦协议流程，先实现一个本地 Mock 服务。

创建payment_gateway/mock_gateway.py：

# 文件路径：payment_gateway/mock_gateway.py import uuid from typing import Dict from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ChargeRequest(BaseModel): invoice_id: str amount: float currency: str class ChargeResponse(BaseModel): receipt_id: str invoice_id: str payment_ref: str amount: float currency: str status: str # 简易内存存储，生产环境应替换为数据库 BALANCE: Dict[str, float] = {"consumer-demo-001": 100.0} @app.post("/v1/charge") def charge(req: ChargeRequest): if req.amount > BALANCE.get("consumer-demo-001", 0): return {"error": "insufficient balance"} BALANCE["consumer-demo-001"] -= req.amount receipt = ChargeResponse( receipt_id=uuid.uuid4().hex, invoice_id=req.invoice_id, payment_ref=f"MOCK-{uuid.uuid4().hex[:8].upper()}", amount=req.amount, currency=req.currency, status="paid" ) return receipt

这里需要特别说明：Mock 支付服务仅用于本地演示。真实生产环境必须满足以下合规要求：

• 资金操作必须经过持牌支付机构，不能自己构建资金池。
• 每一笔交易都要具备完整的审计日志。
• 支付凭据要能支撑后续的对账和退款流程。
• 涉及真实资金的操作必须在测试环境充分验证并备份数据。

4.5 启动服务与验证

我建议用三个终端窗口分别启动三个服务。

终端一，启动 Discovery 与 Provider（示例中二者共用同一个 FastAPI 进程，端口 9001）：

uvicorn provider.main:app --host 0.0.0.0 --port 9001

终端二，启动 Mock 支付网关：

uvicorn payment_gateway.mock_gateway:app --host 0.0.0.0 --port 9003

终端三，运行 Consumer 脚本：

python run_consumer.py

如果一切正常，你会看到类似下面的输出顺序：

发现可用 Agent: [{'agent_id': 'provider-weather-001', 'agent_name': 'Weather Data Agent', ...}] 收到报价: {'proposal_id': '...', 'request_id': '...', 'provider_id': 'provider-weather-001', 'price': 0.01, ...} 执行结果: {'result': {'city': 'Beijing', 'condition': 'cloudy', ...}, 'invoice': {'invoice_id': '...', 'amount': 0.01, ...}} 支付凭证: {'receipt_id': '...', 'invoice_id': '...', 'payment_ref': 'MOCK-...', 'amount': 0.01, 'status': 'paid'}

到了这一步，一个简化版“发现-协商-执行-支付”闭环就完整跑通了。

5. 扩展设计：协商策略与安全机制

5.1 协商不只是“比价”

前面示例中的协商逻辑非常简单，返回的是固定价格。真实场景中，协商需要同时考虑多个因素：

• 请求负载：当 Consumer 传入的参数很大时（比如批量查询 1 万个城市），Provider 的成本远高于单次查询。
• SLA 约束：Consumer 如果要求 50ms 内响应，Provider 可能需要预留更多计算资源，报价也会更高。
• 信任等级：对老客户、高信誉 Consumer，Provider 可以给出折扣或“先服务后付费”的方案。

因此，实际工程中建议把 Proposal 生成逻辑抽象成一个独立的协商模块，规则尽量数据化：

# 文件路径：provider/negotiation_policy.py from protocol.models import ServiceRequest, Proposal, PricingType def build_proposal(req: ServiceRequest) -> Proposal: # 基础价格从数据库或配置读取 base_price = 0.01 # 参数越大，加价越多 if "city_list" in req.parameters: city_num = len(req.parameters["city_list"]) if city_num > 100: base_price += 0.005 * (city_num // 100) # SLA 要求越高，加价越多 if req.required_sla_ms and req.required_sla_ms < 100: base_price += 0.01 return Proposal( request_id=req.request_id, provider_id="provider-weather-001", capability=req.capability, price=round(base_price, 4), currency="USD", estimated_response_ms=req.required_sla_ms or 200, valid_until="2030-12-31T23:59:59Z" )

5.2 身份认证与数据签名

协议开放不等于任何人都能调用。在真实场景中，所有 Agent 之间的消息建议采用 JWT 进行身份认证，同时对关键字段做签名，防止中间人篡改。

例如，Consumer 发来的ServiceRequest中，可以在 header 中携带 JWT：

{ "alg": "RS256", "kid": "consumer-key-1" }

Payload 部分包含：

{ "sub": "consumer-demo-001", "iat": 1699000000, "exp": 1699003600 }

Provider 接收到请求后，先验证 JWT 签名，再校验request_id是否重复，然后才进入协商流程。这样做的好处是：

• 防止攻击者伪造知名 Consumer 身份。
• 保证请求内容的不可抵赖性。
• 便于后续对账时定位具体调用方。

5.3 幂等与去重

Agent 之间通过 HTTP 通信，难免会遇到网络超时和重试。假设 Consumer 发出一次执行请求，Provider 已经执行成功，但响应在传输过程中丢失，Consumer 重试时如果不做幂等处理，服务就会被重复执行，费用也会被重复计算。

解决方案是引入“幂等键”。建议在Agreement中增加idempotency_key字段，并在 Provider 的执行接口中维护一个“已处理请求 ID”表：

import hashlib from fastapi import HTTPException processed_keys = set() def check_idempotency(agreement_id: str): digest = hashlib.sha256(agreement_id.encode()).hexdigest() if digest in processed_keys: raise HTTPException(status_code=409, detail="duplicate agreement") processed_keys.add(digest)

把check_idempotency放在 Providerexecute接口的最前面，可以有效防止重复执行。

5.4 超时与补偿机制

协议中涉及 Provider 和 Consumer 两个状态机。建议为协商和支付增加超时控制：

• Consumer 收到 Proposal 后，有效期由valid_until字段控制，超时后 Proposal 自动失效。
• Consumer 调用支付网关后，如果迟迟未收到 Receipt，应调用支付网关的查询接口确认支付状态，再决定是否执行服务。

这种“先确认，后执行”的策略能够尽量降低 Agent 协作中的资源浪费。

6. 常见问题与排查思路

6.1 Provider 注册后无法被发现

问题现象：Provider 启动成功，但 Consumer 调用发现接口时返回空列表。

排查步骤：

1. 确认 Discovery Server 与 Provider 是否处于同一数据库文件路径。
2. 检查offer.status是否为available。
3. 检查能力名称是否完全一致，例如weather.query多一个空格就无法匹配。
4. 查看 SQLite 表中是否有数据：

sqlite3 agent_registry.db select * from agent_offers;

如果发现数据库为空，说明register没有被触发。检查 FastAPI 的startup事件是否注册正确。

6.2 收到重复执行请求

问题现象：Consumer 因为网络超时重试，导致 Provider 同一服务被执行两次。

解决方案：

1. 在 Provider 执行接口中加入幂等控制，收到重复Agreement时返回409 Conflict。
2. 在 Consumer 端记录当前请求的request_id，重试时复用同一 ID。

6.3 支付回调丢失

问题现象：Consumer 已经扣款，但 Provider 没有收到支付通知。

解决方案：

1. 设计轮询机制：Consumer 每 5 秒查询一次支付状态。
2. 设计 webhook 回调：支付网关在支付成功后主动通知 Provider。
3. 对账兜底：每天定时任务检查未完成的Invoice。

7. 最佳实践与工程建议

7.1 协议版本管理

开放协议一旦被多个团队使用，版本变更就会影响大量 Agent。建议在 URL 路径中加入版本号，例如/v1/negotiate、/v2/negotiate。新增非兼容字段时，优先新增一个新端点，而不是修改现有端点。

7.2 配置管理

每个 Agent 的 Provider 配置建议使用统一配置中心管理，而不是硬编码在代码中。关键配置包括：

• Agent ID 和密钥。
• 注册中心地址。
• 默认报价与折扣规则。
• 支付网关地址。
• 最大请求并发数。

7.3 日志与审计

Agent 协议涉及资金，所有消息交互建议记录完整审计日志，关键字段包括：

• 请求时间戳和响应时间戳。
• 原始ServiceRequest和Proposal全文。
• 支付参考号和账单号。
• 异常分支的堆栈信息。

日志不要打明文密钥，但需要保留消息摘要，便于事后追溯。

7.4 安全边界

• 不允许 Agent 之间直接传递 API Key 或数据库密码。
• Agent 的调用凭证应通过密钥管理服务动态获取。
• 开放协议的鉴权建议使用短时效 Token，支持轮换。
• 涉及内部服务时，Provider 需要做出口访问控制，避免 Agent 被诱导访问内网资源。

7.5 灰度发布与回滚

当 Provider 调整报价策略或模型逻辑时，建议先对部分 Consumer 灰度。可以在协议中增加experimental标志，或者按consumer_id白名单进行路由。一旦发现错误率高或费用异常，立即切换回旧版本 Agent。

8. 总结与下一步学习方向

通过本文的梳理和实战，我们跑通了一条完整的 AI Agent 协作链路：

• 用 Discovery 服务解决“Agent 之间如何发现彼此”的问题；
• 用 Request/Proposal/Agreement 三层消息体解决“如何协商并达成一致”的问题；
• 用 Invoice/Receipt 和 Mock 支付网关解决“如何完成费用结算”的问题。

在此基础上，你可以继续从以下方向深入：

1. 引入消息队列：把同步 HTTP 调用替换为异步消息，提升系统吞吐量和稳定性。
2. 结合大模型：让 Agent 使用大模型自动生成ServiceRequest或自动评估Proposal，实现更高级的自动化决策。
3. 分布式信任体系：研究基于可验证凭证（VC）的 Agent 身份体系，让跨机构的 Agent 协作具备更可靠的安全保障。
4. 真实支付渠道：接入合规支付服务商，完善退款、对账、发票等能力。

开放协议的意义不在于代码本身，而在于它能沉淀出一套标准化的 Agent 协作规则。如果你想在自己的项目中落地，建议从一个最小闭环开始，先跑通再迭代。如果本文对你有帮助，可以收藏备用，后续我会继续分享 Agent 协议与支付集成的更多工程细节。