结果验证、幂等性与错误处理

MCP 工具调用在实际环境中会遇到网络超时、API 限流、部分成功等各种异常情况。本节建立生产级的错误处理和结果验证机制。

MCP 错误处理的三层架构

参数验证：让错误信息有意义

差的错误信息让 LLM 无法自动修复；好的错误信息让 LLM 能够自我纠正：

def validate_and_execute_query(sql: str, db_path: str) -> str:
# ❌ 差的错误处理
# try:
#     conn.execute(sql)
# except Exception as e:
#     raise e  # LLM 收到原始异常，无法理解
# ✅ 好的错误处理：提供上下文和修复建议
sql_lower = sql.strip().lower()
if sql_lower.startswith(("drop", "truncate", "delete from")):
return (
"错误：不允许执行破坏性操作（DROP/TRUNCATE/DELETE）。\n"
"如果需要删除数据，请描述你的目的，我来帮你设计安全的替代方案。"
)
if not sql_lower.startswith("select"):
return (
f"错误：当前数据库连接只支持 SELECT 查询，收到的是 {sql_lower.split()[0].upper()} 语句。\n"
"如需写入数据，请使用 write_query 工具（仅对特定表开放）。"
)
try:
result = execute_select(sql, db_path)
return format_result(result)
except Exception as e:
# 提供具体的错误位置和可能的修复方向
error_msg = str(e)
if "no such table" in error_msg:
tables = get_table_list(db_path)
return f"错误：表不存在。可用的表：{', '.join(tables)}\n原始错误：{error_msg}"
elif "syntax error" in error_msg:
return f"SQL 语法错误：{error_msg}\n建议：检查引号是否配对、列名是否正确。"
else:
return f"执行错误：{error_msg}"

幂等性设计

幂等性（Idempotency）对可能被重试的操作至关重要——相同的操作执行多次，结果应该一致。

非幂等操作的危险：

# ❌ 危险：重试时会创建多条记录
def create_record(data: dict):
cursor.execute("INSERT INTO records VALUES (?)", data)
conn.commit()
# ✅ 安全：使用幂等键，重复调用不产生重复数据
def create_record_idempotent(data: dict, idempotency_key: str):
# 先检查是否已存在
existing = cursor.execute(
"SELECT id FROM records WHERE idempotency_key = ?",
(idempotency_key,)
).fetchone()
if existing:
return {"status": "already_exists", "id": existing[0]}
cursor.execute(
"INSERT INTO records (data, idempotency_key) VALUES (?, ?)",
(json.dumps(data), idempotency_key)
)
conn.commit()
return {"status": "created", "id": cursor.lastrowid}

在 MCP Server 中实现幂等工具：

@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "create_order":
# 使用请求内容的哈希作为幂等键
import hashlib
idempotency_key = hashlib.md5(
json.dumps(arguments, sort_keys=True).encode()
).hexdigest()
return await create_order_idempotent(arguments, idempotency_key)

重试机制

import asyncio
from typing import Callable, Any
async def with_retry(
func: Callable,
max_retries: int = 3,
base_delay: float = 1.0,
backoff_factor: float = 2.0,
retryable_errors: tuple = (TimeoutError, ConnectionError)
) -> Any:
"""带指数退避的重试装饰器"""
last_error = None
for attempt in range(max_retries + 1):
try:
return await func()
except retryable_errors as e:
last_error = e
if attempt < max_retries:
delay = base_delay * (backoff_factor ** attempt)
await asyncio.sleep(delay)
continue
except Exception as e:
# 非重试性错误，立即返回
raise e
# 超过重试次数
return {
"status": "error",
"message": f"操作失败，已重试 {max_retries} 次",
"last_error": str(last_error)
}

API 限流处理

import time
from collections import deque
class RateLimiter:
"""滑动窗口限速器"""
def __init__(self, max_calls: int, period: float):
self.max_calls = max_calls
self.period = period  # 秒
self.calls = deque()
async def acquire(self):
now = time.time()
# 清除过期的调用记录
while self.calls and self.calls[0] < now - self.period:
self.calls.popleft()
if len(self.calls) >= self.max_calls:
# 计算需要等待的时间
wait_time = self.period - (now - self.calls[0])
await asyncio.sleep(wait_time)
self.calls.append(time.time())
# 使用示例（Shopify API 限制：每秒2次）
shopify_limiter = RateLimiter(max_calls=2, period=1.0)
async def call_shopify_api(endpoint: str, params: dict):
await shopify_limiter.acquire()
async with httpx.AsyncClient() as client:
return await client.get(endpoint, params=params, headers=HEADERS)

结果验证

在返回工具结果前，添加基本的合理性检查：

def validate_price_result(price: float, product_name: str) -> dict:
"""验证价格是否合理"""
if price < 0:
return {
"status": "suspicious",
"value": price,
"warning": f"产品 '{product_name}' 的价格为负数（{price}），请人工核查"
}
if price > 100000:
return {
"status": "suspicious",
"value": price,
"warning": f"产品 '{product_name}' 的价格异常高（{price}），可能是单位错误"
}
return {"status": "ok", "value": price}

错误处理最佳实践汇总

本节执行清单

• [ ] 参数验证：非法参数返回描述性错误（告诉 LLM 如何修正）
• [ ] 写操作：使用幂等键，防止重复执行
• [ ] 网络操作：实现有限次数的指数退避重试
• [ ] 结果验证：对关键数字（价格、数量）做合理性检查

下一章：第 07 章 — 构建你自己的 MCP Server