尧图网页设计 
MCP 支付流程里Agent 联系不上 Signer 怎么办一套可落地的降级与重试方案MCPModel Context Protocol现在基本是 Agent 开发绕不开的一层协议。它解决的核心问题是让大模型以标准方式调用外部工具而不是每个应用写一套私有函数调用。但当 MCP 被用在支付流程里时问题就变了Agent 发起支付请求后需要 Signer授权人确认签名可 Signer 不在线、忽略通知、或者审批系统干脆返回不可达这时候 Agent 该做什么是反复重试等到超时还是直接失败让用户重新发起两种方案体验都很差。这次我们来看一个更工程化的思路在 MCP 支付流程里把 “Agent 联系不上 Signer” 当成一个正常状态而不是异常。用状态机管理支付生命周期用异步通知和回调代替同步等待再配合超时、重试、备用 Signer 升级、人工审批通道把支付流程做成“即使 Signer 不在流程也能继续推进而不是卡死”。文章会给出完整的 Python 代码示例包括 MCP Server 端工具定义、Signer 可达性探测、超时处理、降级策略以及测试验证方法。1. 核心能力速览能力项说明项目定位基于 MCP 协议的支付审批流程设计重点解决 Agent 无法联系 Signer 时的状态流转与降级处理协议基础MCPModel Context ProtocolAgent 通过 MCP Server 调用支付相关工具核心功能支付创建、Signer 可达性探测、超时重试、备用 Signer 升级、状态查询、取消支付技术栈Python 3.10mcp 官方 SDKasyncioWebhook 回调启动方式MCP Server 标准输入输出模式可配合 Claude Desktop、Dify 等宿主接入是否支持 API支持MCP Server 本身即工具接口也可额外暴露 RESTful 回调端口是否支持批量任务支持支付请求进入异步队列Signer 可批量审批显存/算力要求无这是纯业务逻辑服务不涉及模型推理适合场景需要人工授权才能执行的支付/审批/发布类 Agent 流程整个方案的重点不是“如何调用 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 环境要求依赖版本说明Python3.10 及以上mcp官方 Python SDK用于实现 MCP Serverhttpx测试 Webhook 推送时使用pytest运行测试用例uvicorn / fastapi如果要把回调接口暴露成 REST API 时使用安装依赖pip install mcp httpx pytest4.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 fPAY-{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 approver2example.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 { approver1example.com: {email: True, webhook: False, push: True}, approver2example.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), timeouttimeout ) 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 fPAY-{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, timeouttimeout_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, timeouttimeout_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, }流程设计要点先等主 Signer超时后再检查可达性。这个顺序很关键——很多实现会先检查可达性再等待但“可达性检查”本身有延迟而且检查结果不能代表后续几分钟 Signer 是否会操作。超时后不直接失败而是升级到备用 Signer。这保证了在团队场景下主审批人休假或被通知淹没时支付不会卡住。升级后再次设置独立的超时时间。不能复用第一次的剩余时间否则备用 Signer 收到通知时只剩几秒注定超时。最终超时后返回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: approver1example.com, fallback_signer_id: approver2example.com, timeout_seconds: 60 } }Server 返回结果示例{ payment_id: PAY-0001, status: expired, error: primary_signer_timeout_and_fallback_timeout, primary_signer: approver1example.com, fallback_signer: approver2example.com }Agent 拿到这个结果后处理策略可以是给用户发送消息“支付已超过审批时限两个审批人都未响应是否需要重新发起”或者自动调整金额拆分为更小金额进入快速审批通道或者将支付状态同步给上游系统标记为需要人工介入。这些决策逻辑要写在 Agent 的 system prompt 或编排层而不是 MCP Server 里。MCP Server 负责提供可靠的工具调用和状态信息Agent 负责根据业务上下文做决策。10. 批量支付多个 Payment 与 Signer 批量审批如果同时有大量支付等待同一个 Signer 审批逐个等待显然不合适。一个更合理的做法是Agent 分批创建支付统一进入AWAITING_SIGNER状态Signer 端收到通知后可以批量查看待审批列表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: fpayment {payment_id} not found} state self.payment_states[payment_id] if state.status ! PaymentStatus.AWAITING_SIGNER: return {error: fpayment {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 APIPOST /callback/signatureBody 包含payment_id、signer_id、approved消息队列Signer 客户端发送审批消息到 MQMCP 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 调用支付工具后长时间无响应同步等待 Signer超时时间设置过长或未设置超时检查 MCP 日志和支付状态改为异步等待使用asyncio.wait_for设置超时Signer 收到通知但支付状态一直是 AWAITING_SIGNER回调接口未接通或验签失败查看回调日志确认签名校验逻辑修复回调地址确认 HMAC 密钥一致提示agent execution terminated due to errorAgent 宿主对单次工具调用有时间限制而等待 Signer 的流程超过了宿主限制查看宿主配置的 tool timeout缩短同步超时或改为“创建支付 轮询状态”两步调用升级到备用 Signer 后备用也一样卡住备用 Signer 的实际在线时间与主 Signer 重合或者在同一个时区同样繁忙查看 Signer 的活跃时间分布设置多级升级链结合排班表选择当前在线的 Signer批量支付时状态丢失使用内存字典存储状态服务重启后丢失查看服务重启时间与支付创建时间引入 Redis/PostgreSQL 持久化回调被重复调用Signer 客户端或 MQ 消费端做了重试投递检查回调日志中同一 payment_id 出现的次数回调接口做幂等处理状态转移时校验当前状态这里特别强调一下agent execution terminated due to error这个问题。模型上下文协议本身没有规定单次工具调用的超时上限但实际的 Agent 宿主比如 Claude Desktop、自研 Agent Runner、Dify通常都有自己的超时控制。一旦 Agent 宿主觉得工具调用时间过长会直接终止执行。解决方案是避免在单次工具调用里等待 Signer 很长时间而是使用“创建-查询”模式Agent 调用request_payment_with_escalation但内部等待时间控制在宿主允许范围内如果超时Server 返回payment_id和当前状态Agent 后续可以多次调用get_payment_status查询最新状态或者使用通知回调让 MCP Server 在状态变化时主动通知 Agent。这样就把“等待 Signer”的压力从 Agent 执行上下文转移到了异步基础设施里。13. 权限限定与审计支付流程的安全要求远高于普通工具调用。几个必须做的点Signer 身份认证与签名记录Signer 操作前必须经过身份认证不能仅凭邮件地址自动信任签名动作要记录时间戳、IP、设备信息和使用的认证方式。Agent 权限最小化Agent 可以创建支付请求但不能直接绕过 Signer 调用支付网关execute_payment工具必须校验状态是SIGNED且校验签名记录是否真实存在。备用 Signer 升级链路升级到备用 Signer 不能由 Agent 自行决定应该是系统自动执行并记录升级原因和触发条件。审计日志支付从创建到完成的每次状态转换都要落日志日志字段建议包含时间、支付单号、操作者Agent or Human、操作类型、旧状态、新状态、上下文信息。上下文泄露风险不要把支付 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 不可达当状态而不是当异常。用支付状态机管理流程用异步通知替代同步等待用超时和升级策略保证流程卡住时能自动推进用审计日志保证每笔支付都有迹可查。值得先验证的三个点是状态机的流转是否符合预期尤其是AWAITING_SIGNER - ESCALATED - EXPIRED这条链路wait_for_signature的超时逻辑能否在指定时间准确触发不占用额外的线程资源MCP 工具返回给 Agent 的状态信息是否足够清晰Agent 是否能根据这些信息做出正确决策。最容易踩的坑是在 MCP Server 内部阻塞等待 Signer导致 Agent 宿主超时终止执行。解决方案就是上面说的异步等待 状态查询 回调通知。后续继续扩展的方向包括接入多级审批链、对接企业微信/钉钉审批流、把支付状态存储迁移到 Redis/PostgreSQL、增加监控告警和 Signer 在线状态预测。如果你们团队正在做 Agent 支付审批这套状态机和降级链路可以直接参考。 MCP支付流程中Signer失联怎么办？一套状态机降级与重试方案
MCP 支付流程里Agent 联系不上 Signer 怎么办一套可落地的降级与重试方案MCPModel Context Protocol现在基本是 Agent 开发绕不开的一层协议。它解决的核心问题是让大模型以标准方式调用外部工具而不是每个应用写一套私有函数调用。但当 MCP 被用在支付流程里时问题就变了Agent 发起支付请求后需要 Signer授权人确认签名可 Signer 不在线、忽略通知、或者审批系统干脆返回不可达这时候 Agent 该做什么是反复重试等到超时还是直接失败让用户重新发起两种方案体验都很差。这次我们来看一个更工程化的思路在 MCP 支付流程里把 “Agent 联系不上 Signer” 当成一个正常状态而不是异常。用状态机管理支付生命周期用异步通知和回调代替同步等待再配合超时、重试、备用 Signer 升级、人工审批通道把支付流程做成“即使 Signer 不在流程也能继续推进而不是卡死”。文章会给出完整的 Python 代码示例包括 MCP Server 端工具定义、Signer 可达性探测、超时处理、降级策略以及测试验证方法。1. 核心能力速览能力项说明项目定位基于 MCP 协议的支付审批流程设计重点解决 Agent 无法联系 Signer 时的状态流转与降级处理协议基础MCPModel Context ProtocolAgent 通过 MCP Server 调用支付相关工具核心功能支付创建、Signer 可达性探测、超时重试、备用 Signer 升级、状态查询、取消支付技术栈Python 3.10mcp 官方 SDKasyncioWebhook 回调启动方式MCP Server 标准输入输出模式可配合 Claude Desktop、Dify 等宿主接入是否支持 API支持MCP Server 本身即工具接口也可额外暴露 RESTful 回调端口是否支持批量任务支持支付请求进入异步队列Signer 可批量审批显存/算力要求无这是纯业务逻辑服务不涉及模型推理适合场景需要人工授权才能执行的支付/审批/发布类 Agent 流程整个方案的重点不是“如何调用 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 环境要求依赖版本说明Python3.10 及以上mcp官方 Python SDK用于实现 MCP Serverhttpx测试 Webhook 推送时使用pytest运行测试用例uvicorn / fastapi如果要把回调接口暴露成 REST API 时使用安装依赖pip install mcp httpx pytest4.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 fPAY-{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 approver2example.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 { approver1example.com: {email: True, webhook: False, push: True}, approver2example.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), timeouttimeout ) 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 fPAY-{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, timeouttimeout_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, timeouttimeout_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, }流程设计要点先等主 Signer超时后再检查可达性。这个顺序很关键——很多实现会先检查可达性再等待但“可达性检查”本身有延迟而且检查结果不能代表后续几分钟 Signer 是否会操作。超时后不直接失败而是升级到备用 Signer。这保证了在团队场景下主审批人休假或被通知淹没时支付不会卡住。升级后再次设置独立的超时时间。不能复用第一次的剩余时间否则备用 Signer 收到通知时只剩几秒注定超时。最终超时后返回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: approver1example.com, fallback_signer_id: approver2example.com, timeout_seconds: 60 } }Server 返回结果示例{ payment_id: PAY-0001, status: expired, error: primary_signer_timeout_and_fallback_timeout, primary_signer: approver1example.com, fallback_signer: approver2example.com }Agent 拿到这个结果后处理策略可以是给用户发送消息“支付已超过审批时限两个审批人都未响应是否需要重新发起”或者自动调整金额拆分为更小金额进入快速审批通道或者将支付状态同步给上游系统标记为需要人工介入。这些决策逻辑要写在 Agent 的 system prompt 或编排层而不是 MCP Server 里。MCP Server 负责提供可靠的工具调用和状态信息Agent 负责根据业务上下文做决策。10. 批量支付多个 Payment 与 Signer 批量审批如果同时有大量支付等待同一个 Signer 审批逐个等待显然不合适。一个更合理的做法是Agent 分批创建支付统一进入AWAITING_SIGNER状态Signer 端收到通知后可以批量查看待审批列表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: fpayment {payment_id} not found} state self.payment_states[payment_id] if state.status ! PaymentStatus.AWAITING_SIGNER: return {error: fpayment {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 APIPOST /callback/signatureBody 包含payment_id、signer_id、approved消息队列Signer 客户端发送审批消息到 MQMCP 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 调用支付工具后长时间无响应同步等待 Signer超时时间设置过长或未设置超时检查 MCP 日志和支付状态改为异步等待使用asyncio.wait_for设置超时Signer 收到通知但支付状态一直是 AWAITING_SIGNER回调接口未接通或验签失败查看回调日志确认签名校验逻辑修复回调地址确认 HMAC 密钥一致提示agent execution terminated due to errorAgent 宿主对单次工具调用有时间限制而等待 Signer 的流程超过了宿主限制查看宿主配置的 tool timeout缩短同步超时或改为“创建支付 轮询状态”两步调用升级到备用 Signer 后备用也一样卡住备用 Signer 的实际在线时间与主 Signer 重合或者在同一个时区同样繁忙查看 Signer 的活跃时间分布设置多级升级链结合排班表选择当前在线的 Signer批量支付时状态丢失使用内存字典存储状态服务重启后丢失查看服务重启时间与支付创建时间引入 Redis/PostgreSQL 持久化回调被重复调用Signer 客户端或 MQ 消费端做了重试投递检查回调日志中同一 payment_id 出现的次数回调接口做幂等处理状态转移时校验当前状态这里特别强调一下agent execution terminated due to error这个问题。模型上下文协议本身没有规定单次工具调用的超时上限但实际的 Agent 宿主比如 Claude Desktop、自研 Agent Runner、Dify通常都有自己的超时控制。一旦 Agent 宿主觉得工具调用时间过长会直接终止执行。解决方案是避免在单次工具调用里等待 Signer 很长时间而是使用“创建-查询”模式Agent 调用request_payment_with_escalation但内部等待时间控制在宿主允许范围内如果超时Server 返回payment_id和当前状态Agent 后续可以多次调用get_payment_status查询最新状态或者使用通知回调让 MCP Server 在状态变化时主动通知 Agent。这样就把“等待 Signer”的压力从 Agent 执行上下文转移到了异步基础设施里。13. 权限限定与审计支付流程的安全要求远高于普通工具调用。几个必须做的点Signer 身份认证与签名记录Signer 操作前必须经过身份认证不能仅凭邮件地址自动信任签名动作要记录时间戳、IP、设备信息和使用的认证方式。Agent 权限最小化Agent 可以创建支付请求但不能直接绕过 Signer 调用支付网关execute_payment工具必须校验状态是SIGNED且校验签名记录是否真实存在。备用 Signer 升级链路升级到备用 Signer 不能由 Agent 自行决定应该是系统自动执行并记录升级原因和触发条件。审计日志支付从创建到完成的每次状态转换都要落日志日志字段建议包含时间、支付单号、操作者Agent or Human、操作类型、旧状态、新状态、上下文信息。上下文泄露风险不要把支付 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 不可达当状态而不是当异常。用支付状态机管理流程用异步通知替代同步等待用超时和升级策略保证流程卡住时能自动推进用审计日志保证每笔支付都有迹可查。值得先验证的三个点是状态机的流转是否符合预期尤其是AWAITING_SIGNER - ESCALATED - EXPIRED这条链路wait_for_signature的超时逻辑能否在指定时间准确触发不占用额外的线程资源MCP 工具返回给 Agent 的状态信息是否足够清晰Agent 是否能根据这些信息做出正确决策。最容易踩的坑是在 MCP Server 内部阻塞等待 Signer导致 Agent 宿主超时终止执行。解决方案就是上面说的异步等待 状态查询 回调通知。后续继续扩展的方向包括接入多级审批链、对接企业微信/钉钉审批流、把支付状态存储迁移到 Redis/PostgreSQL、增加监控告警和 Signer 在线状态预测。如果你们团队正在做 Agent 支付审批这套状态机和降级链路可以直接参考。