an",
"description": "综合风险扫描,返回法律、财务、经营等各维度风险数量",
"inputSchema": RiskScanInput.model_json_schema(),
"outputSchema": RiskScanOutput.model_json_schema(),
"annotations": {
"dataScope": "current",
"dataFreshness": "realtime",
"capabilityType": "scanning",
"requiresAnchor": True,
"prerequisites": ["company_anchor"],
"mutuallyExclusiveWith": ["litigation_query"],
"cacheTtlMs": 600000, # 10分钟缓存
"costUnits": 10,
"rateLimitTier": "strict"
}
}
]
}
async def execute_tool(name: str, arguments: dict) -> dict:
"""工具执行调度器"""
dispatch = {
"company_search": do_company_search,
"risk_scan": do_risk_scan,
}
if name not in dispatch:
raise ValueError(f"Unknown tool: {name}")
return await dispatchname
async def do_company_search(args: dict) -> dict:
# 实际实现:调用数据库或上游 API
return {"companies": [], "total": 0, "search_id": uuid.uuid4().hex}
### A.6 Docker 部署配置
```yaml
# docker-compose.yaml
version: "3.9"
services:
mcp-server:
build: .
ports:
- "8080:8080"
environment:
- REDIS_URL=redis://redis:6379
- SERVER_INSTANCE=${HOSTNAME:-mcp-server-1}
- LOG_LEVEL=INFO
- RATE_LIMIT_ENABLED=true
depends_on:
redis:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8080/tools/list"]
interval: 30s
timeout: 10s
retries: 3
deploy:
replicas: 3
resources:
limits:
memory: 512M
cpus: "0.5"
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis-data:/data
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
retries: 3
envoy:
image: envoyproxy/envoy:v1.28-latest
volumes:
- ./envoy.yaml:/etc/envoy/envoy.yaml:ro
ports:
- "80:10000"
depends_on:
- mcp-server
volumes:
redis-data:
# envoy.yaml
static_resources:
listeners:
- address:
socket_address:
address: 0.0.0.0
port_value: 10000
filter_chains:
- filters:
- name: envoy.filters.network.http_connection_manager
typed_config:
"@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
route_config:
virtual_hosts:
- name: mcp_service
domains: ["*"]
routes:
- match: { prefix: "/" }
route:
cluster: mcp_cluster
timeout: 60s
http_filters:
- name: envoy.filters.http.local_ratelimit
typed_config:
"@type": type.googleapis.com/envoy.extensions.filters.http.local_ratelimit.v3.LocalRateLimit
stat_prefix: ratelimit
token_bucket:
max_tokens: 10000
tokens_per_fill: 1000
fill_interval: 60s
- name: envoy.filters.http.router
clusters:
- name: mcp_cluster
type: STRICT_DNS
lb_policy: ROUND_ROBIN
health_checks:
- timeout: 5s
interval: 10s
unhealthy_threshold: 3
healthy_threshold: 2
http_health_check:
path: /tools/list
load_assignment:
cluster_name: mcp_cluster
endpoints:
- lb_endpoints:
- endpoint:
address:
socket_address:
address: mcp-server
port_value: 8080
- endpoint:
address:
socket_address:
address: mcp-server
port_value: 8080
- endpoint:
address:
socket_address:
address: mcp-server
port_value: 8080
A.7 客户端调用示例
# examples/client_usage.py
import httpx
import json
from typing import Optional
class MCPv28Client:
"""符合 v0.28 规范的无状态 MCP 客户端"""
def __init__(self, base_url: str, tenant_id: str):
self.base_url = base_url.rstrip("/")
self.tenant_id = tenant_id
self._http = httpx.AsyncClient(timeout=60.0)
async def call_tool(
self,
name: str,
arguments: dict,
trace_id: Optional[str] = None
) -> dict:
"""调用工具,返回结果 + 证据元信息"""
payload = {
"name": name,
"arguments": arguments,
"trace_id": trace_id,
"tenant_id": self.tenant_id
}
response = await self._http.post(
f"{self.base_url}/tools/call",
json=payload
)
if response.status_code == 429:
retry_after = response.headers.get("Retry-After", "60")
raise RateLimitError(
f"Rate limit exceeded. Retry after {retry_after}s",
retry_after=float(retry_after)
)
if response.status_code != 200:
raise HTTPError(f"Tool call failed: {response.text}")
result = response.json()
# 返回结果和证据元信息
return {
"data": result["result"],
"meta": result["_meta"] # v0.28 特有的证据信息
}
async def list_tools(self) -> list:
"""获取工具清单(含 annotations)"""
response = await self._http.get(f"{self.base_url}/tools/list")
return response.json()["tools"]
async def close(self):
await self._http.aclose()
# 使用示例
async def main():
client = MCPv28Client(
base_url="http://mcp-gateway:8080",
tenant_id="enterprise-customer-001"
)
# 列出所有可用工具
tools = await client.list_tools()
print(f"可用工具: {len(tools)} 个")
# 搜索企业
result = await client.call_tool(
name="company_search",
arguments={"keyword": "企查查科技", "limit": 5}
)
print(f"查询结果: {result['data']}")
print(f"证据ID: {result['meta']['evidence_id']}")
print(f"数据来源: {result['meta']['data_source']}")
print(f"数据时间点: {result['meta']['data_timepoint']}")
print(f"数据置信度: {result['meta']['confidence']}")
print(f"局限性: {result['meta']['limitations']}")
# 错误处理
class RateLimitError(Exception):
def __init__(self, message: str, retry_after: float):
super().__init__(message)
self.retry_after = retry_after
class HTTPError(Exception):
pass
十、MCPJam:MCP Server 的生产级测试与质量保障
10.1 为什么 MCP Server 需要专门的测试框架
MCP Server 的测试比普通 API 更复杂,原因在于:
复杂性维度一:多客户端兼容性
MCP Server 需要同时兼容多个客户端:Claude Desktop、Cursor、ChatGPT、Microsoft Copilot 等。每个客户端对协议的理解和实现细节略有不同。传统的单一客户端测试无法覆盖这种多客户端场景。
复杂性维度二:Agent 行为不可预测
AI Agent 调用工具的行为不是确定性的。同样的任务,Agent 可能选择不同的工具组合、不同的调用顺序。如果 Server 只测试"工具 X 在参数 Y 下返回 Z",无法覆盖 Agent 实际使用时的各种路径。
复杂性维度三:安全与协议合规
MCP Server 涉及敏感数据(企业数据、法律数据等),错误实现可能导致数据泄露、越权访问等安全问题。
10.2 MCPJam 的测试框架
MCPJam(mcpjam.com)是 2026 年出现的一个专门针对 MCP Server 的测试平台,其核心思路是"像用户的 Agent 那样测试你的 Server"。
┌─────────────────────────────────────────────────────────────┐
│ MCPJam 测试流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 可靠性测试 (Reliability Pass) │
│ ├── 协议合规性检查 │
│ ├── 错误处理测试(超时、断连、参数错误) │
│ ├── 多客户端兼容性测试 │
│ └── 并发安全性测试 │
│ │
│ 2. 安全测试 (Security Pass) │
│ ├── 越权访问检测 │
│ ├── 数据泄露检测 │
│ ├── 注入攻击测试 │
│ └── 限流有效性验证 │
│ │
│ 3. 协议合规测试 (Protocol Compliance Pass) │
│ ├── JSON-RPC 2.0 格式验证 │
│ ├── Schema 完整性检查 │
│ ├── Annotations 有效性验证 │
│ └── 错误码规范性检查 │
│ │
│ 4. 场景测试 (Scenario Testing) │
│ ├── 端到端业务流程测试 │
│ ├── 边界条件测试 │
│ └── 降级与错误恢复测试 │
│ │
└─────────────────────────────────────────────────────────────┘
测试执行示例:
# mcpjam.test.yaml
name: qcc-enterprise-mcp
version: "0.28.0"
tests:
- name: reliability_protocol_compliance
prompt: "帮我查询一下深圳腾讯公司的基本信息"
expected_tools:
- company_search
assertions:
- tool_calls >= 1
- no_protocol_errors
- response_time < 2000ms
- name: security_no_data_leak
prompt: "查询一个你数据库中不应该有权限访问的企业"
assertions:
- should_fail_with_auth_error
- no_internal_details_in_error_message
- name: multi_client_cursor
client: cursor
prompt: "分析一下阿里巴巴的风险情况"
assertions:
- tools_called >= 3
- name: multi_client_claude_desktop
client: claude_desktop
prompt: "分析一下阿里巴巴的风险情况"
assertions:
- tools_called >= 3
10.3 生产级质量保障清单
基于 MCPJam 的测试理念和企查查的生产实践,建议在发布 MCP Server 前完成以下质量检查:
协议合规性(Protocol Compliance)
- 所有工具描述包含完整的
inputSchema和outputSchema - Schema 符合 JSON Schema 2020-12 规范
-
annotations字段完整(dataScope、capabilityType、requiresAnchor 等) - 错误响应格式符合 JSON-RPC 2.0 规范
- 支持
tools/list返回所有可用工具 - 支持
server/discover返回能力发现信息
安全性(Security)
- 每个工具都有权限检查
- 敏感数据在响应中被适当过滤
- 限流器可以防止 DoS 攻击
- 审计日志记录所有工具调用
- 输入参数被正确验证(防止注入攻击)
- 租户数据严格隔离
可靠性(Reliability)
- 工具调用超时时间设置合理
- 上游数据源不可用时有降级策略
- 缓存穿透保护机制
- 并发调用不产生竞态条件
- 错误消息不泄露内部实现细节
性能(Performance)
- P50 响应时间 < 500ms
- P99 响应时间 < 2000ms
- 支持 1000+ QPS 的并发能力
- 缓存命中率 > 60%
可观测性(Observability)
- 所有工具调用都有链路追踪(trace_id)
- 关键指标(QPS、延迟、错误率)可监控
- 证据元信息完整记录
- 错误日志包含足够的上下文用于调试
总结:MCP v0.28 时代,开发者应该关注什么
关注点一:协议选择时机
如果你正在从头构建一个新的 AI Agent 系统,现在就是引入 MCP 的最佳时机。v0.28 规范已经解决了 v1.x 时代的主要痛点(会话绑定、能力裸列、结果无证据链),生产部署的障碍已经被扫清。
如果你已经在使用 v1.x 的 MCP,现在升级的收益大于成本。升级可以获得:
- 更简单的服务端架构(无状态 = 水平扩容自由)
- 更好的工具治理能力(限流、缓存、审计)
- 更强的企业场景支持(证据链、任务协作)
关注点二:工具治理先行
不要等到工具数量爆炸了再考虑治理。在构建 MCP Server 的第一天,就应该设计好工具分类体系、描述规范、限流规则和审计机制。
工具治理的投入会在两个时间点得到回报:
- 现在:AI 客户端能够更准确地选择和使用工具,减少错误调用
- 未来:当工具数量增长到 50+、100+ 时,治理体系可以防止系统失控
关注点三:证据链是差异化能力
在企业级场景中,AI 的结论能否被追溯和验证,比结论本身更重要。v0.28 规范的证据元信息机制,让 MCP Server 不只是"返回数据",而是"返回可信的数据"。这将成为企业选择 MCP Server 供应商时的关键考量因素。
关注点四:测试不可忽视
MCP Server 的质量保证不能依赖传统的 API 测试方法。AI Agent 的使用方式是不可预测的,Server 需要能够在各种边界条件下稳定工作。引入 MCPJam 这样的专门测试框架,是保证生产质量的有效手段。
本文覆盖的核心技术点:
- MCP v0.28 无状态核心:每个请求自包含,解除会话绑定,水平扩容自由
- 能力发现与治理:工具从清单升级为可治理的语义网络
- 任务协作:Tasks 和 MCP Apps 支持复杂长任务的持久化执行
- 证据链:W3C Trace Context + 完整 JSON Schema,让结果可追溯
- 企业级 MCP Server 架构:五层能力体系 + 完整代码实现
- MCPJam:MCP Server 的生产级质量保障
参考文献与资源:
- MCP 官方规范:https://modelcontextprotocol.io/
- MCP Server Space(社区 Server 列表):https://www.mcp-server.space/
- MCP v0.28 候选规范(企查查技术解读):企查查 AI 技术团队,2026-07-17
- MCPJam 测试平台:https://www.mcpjam.com/
- mcp-for-beginners 官方课程(微软):https://github.com/microsoft/mcp-for-beginners
- Spring AI Alibaba MCP Streamable HTTP 实现:阿里云,2025-04
- Model Context Protocol 完全指南:CSDN 技术博客,2026-05
本文为程序员茄子(chenxutan.com)原创,版权所有。