企业官网建设流程全解析

MCP 支付流程里，Agent 联系不上 Signer 怎么办？一套可落地的降级与重试方案

MCP（Model Context Protocol）现在基本是 Agent 开发绕不开的一层协议。它解决的核心问题是：让大模型以标准方式调用外部工具，而不是每个应用写一套私有函数调用。但当 MCP 被用在支付流程里时，问题就变了：Agent 发起支付请求后，需要 Signer（授权人）确认签名，可 Signer 不在线、忽略通知、或者审批系统干脆返回不可达，这时候 Agent 该做什么？是反复重试等到超时？还是直接失败让用户重新发起？两种方案体验都很差。

这次我们来看一个更工程化的思路：在 MCP 支付流程里，把 “Agent 联系不上 Signer” 当成一个正常状态而不是异常。用状态机管理支付生命周期，用异步通知和回调代替同步等待，再配合超时、重试、备用 Signer 升级、人工审批通道，把支付流程做成“即使 Signer 不在，流程也能继续推进，而不是卡死”。文章会给出完整的 Python 代码示例，包括 MCP Server 端工具定义、Signer 可达性探测、超时处理、降级策略以及测试验证方法。

1. 核心能力速览

整个方案的重点不是“如何调用 MCP”，而是“在 MCP 工具层之上，如何设计一个支付状态机，让 Agent 在 Signer 不可达时不至于挂死”。

2. 适用场景与使用边界

这套设计适合以下场景：

• 企业报销、采购付款、合同签署等需要人工审批的 Agent 流程；
• Agent 自动发起支付，但支付前必须经过一个或多个人类授权者的场景；
• 审批人可能在 PC、手机、邮件等多种终端，且不能保证实时在线的场景；
• 需要完整审计链路，每笔支付从创建到完成都要有状态记录的合规场景。

不适合的场景：

• 无需人工干预的纯自动扣款（如订阅续费）——直接用接口就够了，不需要引入 Signer 角色；
• 支付系统本身已经封装好等待轮询逻辑的场景——重复设计会增加复杂度和排障成本；
• 对支付时延要求极高（百毫秒级）的实时扣费场景——人工审批流程天然不适合。

使用边界与合规提醒：

支付流程涉及资金安全和用户授权，下面的代码示例是架构演示，不建议直接放在生产环境不加改造就使用。生产环境必须补充：

• Signer 身份认证（短信验证码、TOTP、生物识别等）；
• 支付金额和收款方白名单校验；
• 所有审批记录落库，保证事后可审计；
• 敏感凭证（API Key、私钥、签名信息）不能明文存储，也不能出现在 Agent 上下文里；
• 涉及企业支付时，确认是否符合财务制度和监管要求。

3. 问题拆解：Agent 为什么联系不上 Signer

先理清楚“Agent 联系不上 Signer”到底包含哪些情况。从技术视角看，至少有三类：

第一类是网络不可达。Signer 的审批客户端离线，或者消息推送服务（邮件、Webhook、短信）返回失败。Agent 发送审批通知后，没有得到任何成功的送达回执。

第二类是超时未响应。通知发出去了，但 Signer 没有在预期时间内操作。这不一定是“不可达”，可能是人还在忙，也可能是通知被误判为垃圾邮件。但站在 Agent 的角度，结果是一样的：在同步等待模式下，它会被卡住。

第三类是审批系统自身故障。比如支付网关返回 5xx、MCP Server 崩溃、数据库连接超时等。这类问题不是 Signer 的问题，但表现出来同样是“流程推进不下去”。

针对这三类情况，支付状态机的设计目标就明确：

• 不允许 Agent 在同步等待中无限挂起；
• 每次状态转移都有明确条件；
• Signer 不可达只是从pending转移到escalation的一个触发条件；
• 整个流程可以通过查询接口随时了解当前状态。

接下来我们就从零搭建这套流程。

4. 环境准备与项目结构

4.1 环境要求

安装依赖：

pip install mcp httpx pytest

4.2 项目目录

mcp-payment-flow/ ├── mcp_server.py # MCP Server 入口，定义支付相关工具 ├── payment_service.py # 支付状态机与业务逻辑 ├── signer_manager.py # Signer 可达性探测与通知 ├── config.py # 超时、重试等配置 └── tests/ └── test_payment_flow.py # 测试用例

5. 状态机设计：支付流程的核心

不用数据库，先用 Python 枚举把支付状态定义出来：

# payment_service.py from enum import Enum class PaymentStatus(str, Enum): PENDING = "pending" # 刚创建，待通知 Signer AWAITING_SIGNER = "awaiting_signer" # 已通知 Signer，等待签名 SIGNED = "signed" # Signer 已签名 CONFIRMED = "confirmed" # 支付确认完成 FAILED = "failed" # 支付执行失败 CANCELLED = "cancelled" # 主动取消 EXPIRED = "expired" # 等待超时 ESCALATED = "escalated" # 已升级到备用 Signer

状态流转规则如下：

class PaymentStateMachine: VALID_TRANSITIONS = { PaymentStatus.PENDING: {PaymentStatus.AWAITING_SIGNER, PaymentStatus.CANCELLED, PaymentStatus.FAILED}, PaymentStatus.AWAITING_SIGNER: {PaymentStatus.SIGNED, PaymentStatus.CANCELLED, PaymentStatus.EXPIRED, PaymentStatus.ESCALATED}, PaymentStatus.ESCALATED: {PaymentStatus.AWAITING_SIGNER, PaymentStatus.SIGNED, PaymentStatus.CANCELLED, PaymentStatus.FAILED}, PaymentStatus.SIGNED: {PaymentStatus.CONFIRMED, PaymentStatus.FAILED}, PaymentStatus.CONFIRMED: set(), PaymentStatus.FAILED: set(), PaymentStatus.CANCELLED: set(), PaymentStatus.EXPIRED: set(), } def __init__(self, payment_id: str): self.payment_id = payment_id self.status = PaymentStatus.PENDING def transition(self, new_status: PaymentStatus) -> bool: if new_status in self.VALID_TRANSITIONS[self.status]: old_status = self.status self.status = new_status print(f"[Payment {self.payment_id}] {old_status.value} -> {new_status.value}") return True print(f"[Payment {self.payment_id}] invalid transition: {self.status.value} -> {new_status.value}") return False

设计关键点：

AWAITING_SIGNER状态下，如果超时，不能直接跳到FAILED，而是跳到EXPIRED或ESCALATED。EXPIRED表示这笔支付因为等待时间过长而作废，ESCALATED表示已经转给备用 Signer。区分这两个状态，是为了在审计时能回答一个问题：这笔钱最终是没人批，还是换了人批。

SIGNED之后不直接等于支付完成。Signer 签名只是授权，资金划转还要调用支付网关。网关返回成功才进入CONFIRMED。这是很多 Agent 支付流程容易踩的坑——把“签名”和“支付成功”混在一起，审计时说不清楚。

6. MCP Server 端工具定义

MCP Server 通过@server.tool()装饰器暴露工具给 Agent。这里定义 6 个工具：

# mcp_server.py import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server from payment_service import PaymentStateMachine, PaymentStatus from signer_manager import SignerManager from config import PAYMENT_TIMEOUT_SECONDS server = Server("mcp-payment-flow") signer_manager = SignerManager() payment_states: dict[str, PaymentStateMachine] = {} def get_or_create_payment_state(payment_id: str) -> PaymentStateMachine: if payment_id not in payment_states: payment_states[payment_id] = PaymentStateMachine(payment_id) return payment_states[payment_id] @server.tool() async def create_payment(amount: float, currency: str, purpose: str, signer_id: str, fallback_signer_id: str = "") -> dict: """创建一笔需要 Signer 授权的支付请求，返回支付单号和初始状态""" payment_id = f"PAY-{len(payment_states) + 1:04d}" state = get_or_create_payment_state(payment_id) await signer_manager.notify_signer(signer_id, payment_id, amount, currency, purpose) state.transition(PaymentStatus.AWAITING_SIGNER) return { "payment_id": payment_id, "status": state.status.value, "amount": amount, "currency": currency, "signer_id": signer_id, "fallback_signer_id": fallback_signer_id, "timeout_seconds": PAYMENT_TIMEOUT_SECONDS, } @server.tool() async def check_signer_availability(signer_id: str) -> dict: """检查 Signer 当前是否可达，返回各通道（email、webhook、push）的状态""" reachable, channels = await signer_manager.check_availability(signer_id) return { "signer_id": signer_id, "reachable": reachable, "channels": channels, } @server.tool() async def get_payment_status(payment_id: str) -> dict: """查询一笔支付当前的完整状态""" state = get_or_create_payment_state(payment_id) return { "payment_id": payment_id, "status": state.status.value, "valid_transitions": [s.value for s in PaymentStateMachine.VALID_TRANSITIONS[state.status]], } @server.tool() async def cancel_payment(payment_id: str, reason: str) -> dict: """取消一笔未完成的支付""" state = get_or_create_payment_state(payment_id) if state.transition(PaymentStatus.CANCELLED): return {"payment_id": payment_id, "status": state.status.value, "reason": reason} return {"payment_id": payment_id, "status": state.status.value, "error": "当前状态不可取消"} @server.tool() async def escalate_to_fallback_signer(payment_id: str) -> dict: """当前 Signer 不可达时，升级到备用 Signer""" state = get_or_create_payment_state(payment_id) if state.status != PaymentStatus.AWAITING_SIGNER: return {"payment_id": payment_id, "error": "只有 AWAITING_SIGNER 状态才能升级"} # 从创建支付时记录的 fallback_signer 获取 fallback_signer_id = "approver2@example.com" await signer_manager.notify_signer(fallback_signer_id, payment_id, 0, "", "ESCALATED") state.transition(PaymentStatus.ESCALATED) return {"payment_id": payment_id, "status": state.status.value, "fallback_signer_id": fallback_signer_id} @server.tool() async def execute_payment(payment_id: str) -> dict: """Signer 已签名后，执行实际支付划转""" state = get_or_create_payment_state(payment_id) if state.status != PaymentStatus.SIGNED: return {"payment_id": payment_id, "error": "只有 SIGNED 状态才能执行支付"} # 模拟调用支付网关 await asyncio.sleep(1) success = True if success: state.transition(PaymentStatus.CONFIRMED) return {"payment_id": payment_id, "status": state.status.value, "executed": True} else: state.transition(PaymentStatus.FAILED) return {"payment_id": payment_id, "status": state.status.value, "executed": False}

注意这里escale_to_fallback_signer里 fallback_signer_id 是写死的，实际项目应该在create_payment时把备用 Signer 信息存入数据库或内存字典，这里为了示例简洁先简化。

7. Signer 可达性探测与通知

SignerManager负责两个事情：通知 Signer 有新支付等待审批；探测 Signer 是否可达。

# signer_manager.py import asyncio from typing import Optional class SignerManager: def __init__(self): # 模拟 Signer 在线状态表 self._signer_status = { "approver1@example.com": {"email": True, "webhook": False, "push": True}, "approver2@example.com": {"email": True, "webhook": True, "push": True}, } async def notify_signer(self, signer_id: str, payment_id: str, amount: float, currency: str, purpose: str) -> bool: """发送审批通知""" # 这里用 sleep 模拟一次真实的 HTTP 调用或推送服务调用 await asyncio.sleep(0.2) print(f"[Notify] {signer_id} - payment {payment_id} amount={amount} {currency} purpose={purpose}") return True async def check_availability(self, signer_id: str) -> tuple[bool, dict]: """分段探测 Signer 的可达性""" await asyncio.sleep(0.1) channels = self._signer_status.get(signer_id, {}) reachable = any(channels.values()) return reachable, channels async def wait_for_signature(self, payment_id: str, timeout: int) -> Optional[str]: """ 异步等待 Signer 签名。这里不阻塞当前进程，而是返回一个 Future。 实际项目中可以把它接入消息队列、WebSocket 或轮询数据库。 """ try: return await asyncio.wait_for( self._wait_signer_action(payment_id), timeout=timeout ) except asyncio.TimeoutError: return None async def _wait_signer_action(self, payment_id: str) -> str: # 实际项目中，这里应该监听审批回调事件 # 示例代码用一个循环模拟 await asyncio.sleep(100) return "signed"

这里的关键设计：wait_for_signature是异步等待，不是同步阻塞。asyncio.wait_for会等一个超时时间，超时后抛出TimeoutError，上层根据这个信号决定是重试、升级还是让支付过期。

8. 核心逻辑：Agent 联系不上 Signer 时的完整处理流程

现在到了文章最核心的部分。假设 Agent 调用create_payment创建了一笔支付，Signer 收到通知后一直没有操作。Agent 在 MCP 工具层应该怎么处理？

我们不希望 Agent 自己写一堆繁琐的超时重试代码——那是 MCP Server 的职责。Server 端应该提供一个更高层的工具，比如request_payment_with_escalation，把“通知 Signer、等待签名、超时、升级、再次等待、最终超时”这个完整流程封装起来。Agent 只需要一次调用，就能拿到最终结果。

# mcp_server.py 中新增 @server.tool() async def request_payment_with_escalation( amount: float, currency: str, purpose: str, signer_id: str, fallback_signer_id: str, timeout_seconds: int = 30, ) -> dict: """ 完整的支付审批流程： 1. 创建支付，通知 Signer 2. 异步等待签名，超时后检查 Signer 可达性 3. Signer 不可达则升级到备用 Signer 4. 备用 Signer 再次超时则返回 EXPIRED """ payment_id = f"PAY-{len(payment_states) + 1:04d}" state = get_or_create_payment_state(payment_id) state.transition(PaymentStatus.AWAITING_SIGNER) # 第一轮等待 deadline_channel = await signer_manager.wait_for_signature(payment_id, timeout=timeout_seconds) if deadline_channel: state.transition(PaymentStatus.SIGNED) return {"payment_id": payment_id, "status": state.status.value, "signed_by": signer_id} # 第一轮超时，检查 Signer 可达性 reachable, channels = await signer_manager.check_availability(signer_id) # 即使 Signer 可达但没有操作，也把渠道信息作为审计上下文记录下来 # 升级到备用 Signer state.transition(PaymentStatus.ESCALATED) await signer_manager.notify_signer(fallback_signer_id, payment_id, amount, currency, purpose) # 第二轮等待 deadline_channel_fallback = await signer_manager.wait_for_signature(payment_id, timeout=timeout_seconds) if deadline_channel_fallback: state.transition(PaymentStatus.SIGNED) return {"payment_id": payment_id, "status": state.status.value, "signed_by": fallback_signer_id} # 备用 Signer 也超时，标记过期 state.transition(PaymentStatus.EXPIRED) return { "payment_id": payment_id, "status": state.status.value, "error": "primary_signer_timeout_and_fallback_timeout", "primary_signer": signer_id, "fallback_signer": fallback_signer_id, }

流程设计要点：

1. 先等主 Signer，超时后再检查可达性。这个顺序很关键——很多实现会先检查可达性再等待，但“可达性检查”本身有延迟，而且检查结果不能代表后续几分钟 Signer 是否会操作。

2. 超时后不直接失败，而是升级到备用 Signer。这保证了在团队场景下，主审批人休假或被通知淹没时，支付不会卡住。

3. 升级后再次设置独立的超时时间。不能复用第一次的剩余时间，否则备用 Signer 收到通知时只剩几秒，注定超时。

4. 最终超时后返回EXPIRED状态和错误原因。Agent 拿到这个结果，可以明确告诉用户“这笔支付因为审批超时已过期，请重新发起”，而不是含糊地报错。

这里需要注意的是，上面的示例代码中wait_for_signature内部是 sleep 100 秒，实际项目应该实现为：

• 监听消息队列（如 Redis Stream、RabbitMQ、Kafka）；
• 或者轮询数据库中的审批记录；
• 或者接收 Webhook 回调。

不管哪种方式，对外暴露的接口应该保持同样签名，方便替换实现。

9. Agent 端调用示例

MCP Server 写好后，Agent 如何调用？以 Claude Desktop 或其他 MCP 客户端为例，调用方式大致是：

{ "tool": "request_payment_with_escalation", "arguments": { "amount": 1299.00, "currency": "CNY", "purpose": "Cloud hosting invoice #2025-0331", "signer_id": "approver1@example.com", "fallback_signer_id": "approver2@example.com", "timeout_seconds": 60 } }

Server 返回结果示例：

{ "payment_id": "PAY-0001", "status": "expired", "error": "primary_signer_timeout_and_fallback_timeout", "primary_signer": "approver1@example.com", "fallback_signer": "approver2@example.com" }

Agent 拿到这个结果后，处理策略可以是：

• 给用户发送消息：“支付已超过审批时限，两个审批人都未响应，是否需要重新发起？”
• 或者自动调整金额，拆分为更小金额，进入快速审批通道；
• 或者将支付状态同步给上游系统，标记为需要人工介入。

这些决策逻辑要写在 Agent 的 system prompt 或编排层，而不是 MCP Server 里。MCP Server 负责提供可靠的工具调用和状态信息，Agent 负责根据业务上下文做决策。

10. 批量支付：多个 Payment 与 Signer 批量审批

如果同时有大量支付等待同一个 Signer 审批，逐个等待显然不合适。一个更合理的做法是：

1. Agent 分批创建支付，统一进入AWAITING_SIGNER状态；
2. Signer 端收到通知后，可以批量查看待审批列表；
3. Signer 批量签名后，回调接口统一更新多个支付状态。

在SignerManager中增加一个回调处理函数：

# signer_manager.py 扩展 class PaymentCallbackHandler: def __init__(self, payment_states: dict): self.payment_states = payment_states async def handle_signature_callback(self, payment_id: str, signer_id: str, approved: bool) -> dict: """Signer 客户端回调，标记一笔支付已签名""" from payment_service import PaymentStatus if payment_id not in self.payment_states: return {"error": f"payment {payment_id} not found"} state = self.payment_states[payment_id] if state.status != PaymentStatus.AWAITING_SIGNER: return {"error": f"payment {payment_id} is not awaiting signer"} if approved: state.transition(PaymentStatus.SIGNED) return {"payment_id": payment_id, "status": state.status.value} else: state.transition(PaymentStatus.CANCELLED) return {"payment_id": payment_id, "status": state.status.value}

调用这个回调的方式可以是：

• REST API：POST /callback/signature，Body 包含payment_id、signer_id、approved；
• 消息队列：Signer 客户端发送审批消息到 MQ，MCP Server 消费并更新状态；
• 数据库更新：审批记录写入数据库，Server 定时轮询或触发旁路事件。

要注意回调数据必须验签。不能让任何人随便 POST 一个/callback/signature就把支付标记为“已签名”。生产环境至少要校验签名或用 HMAC 验证请求来自可信来源。

11. 显存占用与资源观察：本方案的真实开销

这个方案不涉及任何模型推理，所以没有显存压力。部署时主要观察的是：

• MCP Server 进程的内存占用（通常在几十 MB 到几百 MB 之间，取决于是否加载了大型依赖）；
• 等待中的异步任务数量；
• 回调接口的并发压力；
• Signer 通知服务的投递延迟。

可以使用如下方式观察服务状态：

# 查看 MCP Server 进程的内存和 CPU ps aux | grep mcp_server.py # Linux 下用 top 或 htop 实时观察 htop

如果支付量很大，建议：

• 把支付状态存储落到 Redis 或 PostgreSQL，而不是内存字典；
• 使用后台 Worker 处理通知投递和超时扫描；
• MCP Server 本身保持轻量，不做重业务；
• 超时任务用定时器扫描数据库，而不是每个支付开一个 asyncio task。

12. 常见问题与排查方法

这里特别强调一下agent execution terminated due to error这个问题。模型上下文协议本身没有规定单次工具调用的超时上限，但实际的 Agent 宿主（比如 Claude Desktop、自研 Agent Runner、Dify）通常都有自己的超时控制。一旦 Agent 宿主觉得工具调用时间过长，会直接终止执行。解决方案是避免在单次工具调用里等待 Signer 很长时间，而是使用“创建-查询”模式：

1. Agent 调用request_payment_with_escalation，但内部等待时间控制在宿主允许范围内；
2. 如果超时，Server 返回payment_id和当前状态；
3. Agent 后续可以多次调用get_payment_status查询最新状态；
4. 或者使用通知回调，让 MCP Server 在状态变化时主动通知 Agent。

这样就把“等待 Signer”的压力从 Agent 执行上下文转移到了异步基础设施里。

13. 权限限定与审计

支付流程的安全要求远高于普通工具调用。几个必须做的点：

1. Signer 身份认证与签名记录

   • Signer 操作前必须经过身份认证，不能仅凭邮件地址自动信任；
   • 签名动作要记录时间戳、IP、设备信息和使用的认证方式。

2. Agent 权限最小化

   • Agent 可以创建支付请求，但不能直接绕过 Signer 调用支付网关；
   • execute_payment工具必须校验状态是SIGNED，且校验签名记录是否真实存在。

3. 备用 Signer 升级链路

   • 升级到备用 Signer 不能由 Agent 自行决定；
   • 应该是系统自动执行，并记录升级原因和触发条件。

4. 审计日志

   • 支付从创建到完成的每次状态转换都要落日志；
   • 日志字段建议包含：时间、支付单号、操作者（Agent or Human）、操作类型、旧状态、新状态、上下文信息。

5. 上下文泄露风险

   • 不要把支付 ID、签名 URL、审批链接直接放进 Agent 的可见上下文里，尤其是需要保密的企业支付信息；
   • MCP 工具返回给 Agent 的信息要严格裁剪，Agent 只需要知道状态和下一步动作，不需要知道完整签名密钥或回调 URL。

14. 最佳实践建议

整套方案跑通之后，工程化落地时可以参考以下建议：

第一，第一次接入先小流量验证。不要直接接生产支付网关，先用一个 mock 支付服务，模拟 Signer 可达、不可达、超时、拒绝四种情况，确认状态机在每种情况下都符合预期。

第二，保留一套最小可运行配置。比如只保留create_payment、get_payment_status、execute_payment三个工具，配合一个写死的wait_for_signature假实现，先把链路打通，再逐步加复杂逻辑。

第三，支付状态、部门、Signer 配置文件分离。状态机代码里不要写死审批人邮箱和部门规则，而是用配置文件或数据库管理：

# config.yaml signers: default_timeout_seconds: 60 max_retries: 3 fallback_policy: escalate payment: currency: CNY min_amount: 0.01 max_amount: 50000 require_second_approval: true

第四，批量任务加日志和失败重试。批量支付和单个支付的处理逻辑要区分开。单个支付可以同步一点，批量支付必须走异步队列，每个任务有独立的任务 ID，方便失败后单独重试。

第五，接入真实环境前做安全评审。确认 Signer 回调验签、日志脱敏、网络访问控制、密钥管理方案都通过评审。

15. 总结

整个方案的核心思路可以压缩成一句话：把 Signer 不可达当状态，而不是当异常。用支付状态机管理流程，用异步通知替代同步等待，用超时和升级策略保证流程卡住时能自动推进，用审计日志保证每笔支付都有迹可查。

值得先验证的三个点是：

1. 状态机的流转是否符合预期，尤其是AWAITING_SIGNER -> ESCALATED -> EXPIRED这条链路；
2. wait_for_signature的超时逻辑能否在指定时间准确触发，不占用额外的线程资源；
3. MCP 工具返回给 Agent 的状态信息是否足够清晰，Agent 是否能根据这些信息做出正确决策。

最容易踩的坑是：在 MCP Server 内部阻塞等待 Signer，导致 Agent 宿主超时终止执行。解决方案就是上面说的异步等待 + 状态查询 + 回调通知。

后续继续扩展的方向包括：接入多级审批链、对接企业微信/钉钉审批流、把支付状态存储迁移到 Redis/PostgreSQL、增加监控告警和 Signer 在线状态预测。如果你们团队正在做 Agent 支付审批，这套状态机和降级链路可以直接参考。