各位读者朋友大家好。最近 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 系统的接口地址、接口参数、认证方式、计费规则甚至还需要单独约定对账方案。这种方式的缺点非常明显无法动态发现A 系统无法在运行时感知 B 系统是否在线、是否支持新的能力。协商成本高价格、响应时间、服务级别SLA通常是写死的无法根据请求上下文动态调整。支付链路断裂Agent 无法自主完成“服务产生费用 - 费用确认 - 资金转移”的整条链路往往需要人工介入。开放协议通过标准化的消息结构、握手流程和结算接口把上述问题抽象成了三个层面层面核心职责类比Discovery发现Agent 如何注册自己的能力如何被其他 Agent 找到服务注册中心 能力目录Negotiation协商Agent 如何交换需求、报价、服务质量约束并达成一致商务谈判 合约签署Payment支付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 核心术语定义在设计协议前先统一几个术语术语含义Agent Provider提供某种能力的 Agent类似服务端Agent Consumer消费某种能力的 Agent类似客户端Offer服务提供方发布的“能力价格”描述Request for ServiceRFS消费方发出的需求描述Proposal提供方针对 RFS 给出的具体报价提案Agreement双方确认后的服务合约Invoice服务完成后的账单Receipt支付完成后的凭证举个例子Consumer 发出一个 RFS内容是“我需要北京市今天下午 3 点到 5 点的降雨概率要求 JSON 格式”。Provider 收到后返回 Proposal内容包括“可以提供该数据单价 0.01 元/次预计响应时间 200ms”。Consumer 确认后生成 AgreementProvider 完成查询后发出 InvoiceConsumer 根据 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.md3.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_factorylambda: 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_factorylambda: 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_factorylambda: 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_factorylambda: 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_factorylambda: uuid4().hex) invoice_id: str payment_ref: str amount: float currency: str status: str paid这里要注意uuid4().hex需要在文件顶部导入uuidfrom 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 statusavailable).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 有两个职责启动时注册自身能力。处理 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_idPROVIDER_ID, agent_nameWeather Data Agent, capabilities[ AgentCapability( nameweather.query, description查询指定城市在指定日期的天气情况, input_schema{ type: object, properties: { city: {type: string}, date: {type: string} }, required: [city] } ) ], pricing_typeper_call, price0.01, currencyUSD, endpointhttp://localhost:9001 ) discovery.register(offer) # 接收消费方的服务请求RFS app.post(/v1/negotiate) def negotiate(req: ServiceRequest): 处理 Consumer 发来的协商请求。 这里简化了协商策略直接返回固定报价。 proposal Proposal( request_idreq.request_id, provider_idPROVIDER_ID, capabilityreq.capability, price0.01, currencyUSD, estimated_response_ms80, valid_until2030-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_idagreement.agreement_id, provider_idagreement.provider_id, consumer_idagreement.consumer_id, amountagreement.price, currencyagreement.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 是整个流程的发起方。它需要完成以下步骤调用发现服务找到满足能力要求的 Provider。向 Provider 发出ServiceRequest。接收Proposal判断价格是否在预算内。生成Agreement并与 Provider 确认。接收Invoice。调用支付网关完成支付获取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(timeout10.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, jsonreq.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, jsonagreement.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, jsonpayload) 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(timeout10.0) # 1. 发现 agents discover_agents(client, weather.query) print(发现可用 Agent:, agents) if not agents: print(没有找到可用服务) return agent agents[0] # 2. 协商 req ServiceRequest( consumer_idconsumer-demo-001, capabilityweather.query, parameters{city: Beijing}, max_budget0.05, required_sla_ms500 ) 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_idreq.request_id, proposal_idproposal[proposal_id], provider_idproposal[provider_id], consumer_idreq.consumer_id, capabilityreq.capability, priceproposal[price], currencyproposal[currency], statusactive ) # 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_iduuid.uuid4().hex, invoice_idreq.invoice_id, payment_reffMOCK-{uuid.uuid4().hex[:8].upper()}, amountreq.amount, currencyreq.currency, statuspaid ) return receipt这里需要特别说明Mock 支付服务仅用于本地演示。真实生产环境必须满足以下合规要求资金操作必须经过持牌支付机构不能自己构建资金池。每一笔交易都要具备完整的审计日志。支付凭据要能支撑后续的对账和退款流程。涉及真实资金的操作必须在测试环境充分验证并备份数据。4.5 启动服务与验证我建议用三个终端窗口分别启动三个服务。终端一启动 Discovery 与 Provider示例中二者共用同一个 FastAPI 进程端口 9001uvicorn 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 可能需要预留更多计算资源报价也会更高。信任等级对老客户、高信誉 ConsumerProvider 可以给出折扣或“先服务后付费”的方案。因此实际工程中建议把 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_idreq.request_id, provider_idprovider-weather-001, capabilityreq.capability, priceround(base_price, 4), currencyUSD, estimated_response_msreq.required_sla_ms or 200, valid_until2030-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_code409, detailduplicate 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 调用发现接口时返回空列表。排查步骤确认 Discovery Server 与 Provider 是否处于同一数据库文件路径。检查offer.status是否为available。检查能力名称是否完全一致例如weather.query多一个空格就无法匹配。查看 SQLite 表中是否有数据sqlite3 agent_registry.db select * from agent_offers;如果发现数据库为空说明register没有被触发。检查 FastAPI 的startup事件是否注册正确。6.2 收到重复执行请求问题现象Consumer 因为网络超时重试导致 Provider 同一服务被执行两次。解决方案在 Provider 执行接口中加入幂等控制收到重复Agreement时返回409 Conflict。在 Consumer 端记录当前请求的request_id重试时复用同一 ID。问题现象常见原因解决思路执行重复HTTP 超时后重试Provider 使用幂等键去重账单金额不符协商价与执行价不一致执行阶段重新校验 AgreementProposal 过期Consumer 处理缓慢校验 valid_until 字段支付失败余额不足或支付渠道异常查询支付网关状态并补偿消息字段校验失败Pydantic 版本差异统一数据模型并升级到 pydantic v26.3 支付回调丢失问题现象Consumer 已经扣款但 Provider 没有收到支付通知。解决方案设计轮询机制Consumer 每 5 秒查询一次支付状态。设计 webhook 回调支付网关在支付成功后主动通知 Provider。对账兜底每天定时任务检查未完成的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 支付网关解决“如何完成费用结算”的问题。在此基础上你可以继续从以下方向深入引入消息队列把同步 HTTP 调用替换为异步消息提升系统吞吐量和稳定性。结合大模型让 Agent 使用大模型自动生成ServiceRequest或自动评估Proposal实现更高级的自动化决策。分布式信任体系研究基于可验证凭证VC的 Agent 身份体系让跨机构的 Agent 协作具备更可靠的安全保障。真实支付渠道接入合规支付服务商完善退款、对账、发票等能力。开放协议的意义不在于代码本身而在于它能沉淀出一套标准化的 Agent 协作规则。如果你想在自己的项目中落地建议从一个最小闭环开始先跑通再迭代。如果本文对你有帮助可以收藏备用后续我会继续分享 Agent 协议与支付集成的更多工程细节。 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 系统的接口地址、接口参数、认证方式、计费规则甚至还需要单独约定对账方案。这种方式的缺点非常明显无法动态发现A 系统无法在运行时感知 B 系统是否在线、是否支持新的能力。协商成本高价格、响应时间、服务级别SLA通常是写死的无法根据请求上下文动态调整。支付链路断裂Agent 无法自主完成“服务产生费用 - 费用确认 - 资金转移”的整条链路往往需要人工介入。开放协议通过标准化的消息结构、握手流程和结算接口把上述问题抽象成了三个层面层面核心职责类比Discovery发现Agent 如何注册自己的能力如何被其他 Agent 找到服务注册中心 能力目录Negotiation协商Agent 如何交换需求、报价、服务质量约束并达成一致商务谈判 合约签署Payment支付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 核心术语定义在设计协议前先统一几个术语术语含义Agent Provider提供某种能力的 Agent类似服务端Agent Consumer消费某种能力的 Agent类似客户端Offer服务提供方发布的“能力价格”描述Request for ServiceRFS消费方发出的需求描述Proposal提供方针对 RFS 给出的具体报价提案Agreement双方确认后的服务合约Invoice服务完成后的账单Receipt支付完成后的凭证举个例子Consumer 发出一个 RFS内容是“我需要北京市今天下午 3 点到 5 点的降雨概率要求 JSON 格式”。Provider 收到后返回 Proposal内容包括“可以提供该数据单价 0.01 元/次预计响应时间 200ms”。Consumer 确认后生成 AgreementProvider 完成查询后发出 InvoiceConsumer 根据 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.md3.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_factorylambda: 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_factorylambda: 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_factorylambda: 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_factorylambda: 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_factorylambda: uuid4().hex) invoice_id: str payment_ref: str amount: float currency: str status: str paid这里要注意uuid4().hex需要在文件顶部导入uuidfrom 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 statusavailable).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 有两个职责启动时注册自身能力。处理 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_idPROVIDER_ID, agent_nameWeather Data Agent, capabilities[ AgentCapability( nameweather.query, description查询指定城市在指定日期的天气情况, input_schema{ type: object, properties: { city: {type: string}, date: {type: string} }, required: [city] } ) ], pricing_typeper_call, price0.01, currencyUSD, endpointhttp://localhost:9001 ) discovery.register(offer) # 接收消费方的服务请求RFS app.post(/v1/negotiate) def negotiate(req: ServiceRequest): 处理 Consumer 发来的协商请求。 这里简化了协商策略直接返回固定报价。 proposal Proposal( request_idreq.request_id, provider_idPROVIDER_ID, capabilityreq.capability, price0.01, currencyUSD, estimated_response_ms80, valid_until2030-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_idagreement.agreement_id, provider_idagreement.provider_id, consumer_idagreement.consumer_id, amountagreement.price, currencyagreement.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 是整个流程的发起方。它需要完成以下步骤调用发现服务找到满足能力要求的 Provider。向 Provider 发出ServiceRequest。接收Proposal判断价格是否在预算内。生成Agreement并与 Provider 确认。接收Invoice。调用支付网关完成支付获取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(timeout10.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, jsonreq.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, jsonagreement.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, jsonpayload) 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(timeout10.0) # 1. 发现 agents discover_agents(client, weather.query) print(发现可用 Agent:, agents) if not agents: print(没有找到可用服务) return agent agents[0] # 2. 协商 req ServiceRequest( consumer_idconsumer-demo-001, capabilityweather.query, parameters{city: Beijing}, max_budget0.05, required_sla_ms500 ) 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_idreq.request_id, proposal_idproposal[proposal_id], provider_idproposal[provider_id], consumer_idreq.consumer_id, capabilityreq.capability, priceproposal[price], currencyproposal[currency], statusactive ) # 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_iduuid.uuid4().hex, invoice_idreq.invoice_id, payment_reffMOCK-{uuid.uuid4().hex[:8].upper()}, amountreq.amount, currencyreq.currency, statuspaid ) return receipt这里需要特别说明Mock 支付服务仅用于本地演示。真实生产环境必须满足以下合规要求资金操作必须经过持牌支付机构不能自己构建资金池。每一笔交易都要具备完整的审计日志。支付凭据要能支撑后续的对账和退款流程。涉及真实资金的操作必须在测试环境充分验证并备份数据。4.5 启动服务与验证我建议用三个终端窗口分别启动三个服务。终端一启动 Discovery 与 Provider示例中二者共用同一个 FastAPI 进程端口 9001uvicorn 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 可能需要预留更多计算资源报价也会更高。信任等级对老客户、高信誉 ConsumerProvider 可以给出折扣或“先服务后付费”的方案。因此实际工程中建议把 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_idreq.request_id, provider_idprovider-weather-001, capabilityreq.capability, priceround(base_price, 4), currencyUSD, estimated_response_msreq.required_sla_ms or 200, valid_until2030-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_code409, detailduplicate 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 调用发现接口时返回空列表。排查步骤确认 Discovery Server 与 Provider 是否处于同一数据库文件路径。检查offer.status是否为available。检查能力名称是否完全一致例如weather.query多一个空格就无法匹配。查看 SQLite 表中是否有数据sqlite3 agent_registry.db select * from agent_offers;如果发现数据库为空说明register没有被触发。检查 FastAPI 的startup事件是否注册正确。6.2 收到重复执行请求问题现象Consumer 因为网络超时重试导致 Provider 同一服务被执行两次。解决方案在 Provider 执行接口中加入幂等控制收到重复Agreement时返回409 Conflict。在 Consumer 端记录当前请求的request_id重试时复用同一 ID。问题现象常见原因解决思路执行重复HTTP 超时后重试Provider 使用幂等键去重账单金额不符协商价与执行价不一致执行阶段重新校验 AgreementProposal 过期Consumer 处理缓慢校验 valid_until 字段支付失败余额不足或支付渠道异常查询支付网关状态并补偿消息字段校验失败Pydantic 版本差异统一数据模型并升级到 pydantic v26.3 支付回调丢失问题现象Consumer 已经扣款但 Provider 没有收到支付通知。解决方案设计轮询机制Consumer 每 5 秒查询一次支付状态。设计 webhook 回调支付网关在支付成功后主动通知 Provider。对账兜底每天定时任务检查未完成的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 支付网关解决“如何完成费用结算”的问题。在此基础上你可以继续从以下方向深入引入消息队列把同步 HTTP 调用替换为异步消息提升系统吞吐量和稳定性。结合大模型让 Agent 使用大模型自动生成ServiceRequest或自动评估Proposal实现更高级的自动化决策。分布式信任体系研究基于可验证凭证VC的 Agent 身份体系让跨机构的 Agent 协作具备更可靠的安全保障。真实支付渠道接入合规支付服务商完善退款、对账、发票等能力。开放协议的意义不在于代码本身而在于它能沉淀出一套标准化的 Agent 协作规则。如果你想在自己的项目中落地建议从一个最小闭环开始先跑通再迭代。如果本文对你有帮助可以收藏备用后续我会继续分享 Agent 协议与支付集成的更多工程细节。