附：MCP Server 实战开发——从零构建企业级 MCP Server
A.1 项目结构与依赖
让我们通过一个完整示例来展示如何构建一个符合 v0.28 规范的生产级 MCP Server。
# 项目结构
mcp-enterprise-server/
├── src/
│ ├── __init__.py
│ ├── server.py # 主服务器入口
│ ├── tools/ # 工具实现
│ │ ├── __init__.py
│ │ ├── business.py # 企业数据工具
│ │ ├── risk.py # 风险查询工具
│ │ └── legal.py # 法律数据工具
│ ├── governance/ # 治理层
│ │ ├── __init__.py
│ │ ├── rate_limiter.py # 限流器
│ │ ├── cost_tracker.py # 成本计量
│ │ └── audit_logger.py # 审计日志
│ ├── schemas/ # Schema 定义
│ │ ├── business.py
│ │ └── risk.py
│ └── utils/
│ ├── evidence.py # 证据元信息生成
│ └── trace.py # 链路追踪
├── pyproject.toml
├── Dockerfile
└── docker-compose.yaml
# pyproject.toml
[project]
name = "mcp-enterprise-server"
version = "0.28.0"
requires-python = ">=3.11"
dependencies = [
"mcp>=1.0.0",
"fastapi>=0.110.0",
"uvicorn>=0.27.0",
"redis>=5.0.0",
"aioredis>=2.0.0",
"httpx>=0.27.0",
"pydantic>=2.6.0",
"structlog>=24.0.0",
"opentelemetry-api>=1.22.0",
"opentelemetry-sdk>=1.22.0",
"opentelemetry-instrumentation-fastapi>=0.43b0",
]
A.2 工具定义：完整的 Schema 与 Evidence
这是 v0.28 规范下工具定义的完整示例：
# src/schemas/business.py
from pydantic import BaseModel, Field
from typing import Optional, List, Literal
from datetime import datetime
class CompanySearchInput(BaseModel):
keyword: str = Field(
description="企业名称关键词（支持模糊匹配）",
min_length=2,
max_length=100
)
data_scope: Literal["current", "history", "all"] = Field(
default="current",
description="数据范围：current=当前在营，history=历史记录，all=全部"
)
province: Optional[str] = Field(
default=None,
description="省份筛选（行政区划代码，如 110000）"
)
limit: int = Field(
default=10,
ge=1,
le=100,
description="返回结果数量上限"
)
class CompanySearchOutput(BaseModel):
companies: List[dict] = Field(
description="匹配的企业列表"
)
total: int = Field(description="符合条件的总企业数")
search_id: str = Field(description="本次查询的唯一标识，用于关联后续查询")
class CompanySearchEvidence(BaseModel):
"""证据元信息：每个工具返回结果必须附带"""
evidence_id: str
trace_id: str
data_source: str
query_timestamp: datetime
data_timepoint: str # "current" | "history"
confidence: float
limitations: List[str]
cache_hit: bool
server_instance: str
A.3 限流器实现
# src/governance/rate_limiter.py
import time
import hashlib
from typing import Dict, Tuple
from collections import defaultdict
from dataclasses import dataclass
import structlog
logger = structlog.get_logger()
@dataclass
class RateLimitRule:
"""限流规则"""
requests_per_minute: int
requests_per_hour: int
burst_size: int # 允许的突发请求数
class TieredRateLimiter:
"""分层限流器：支持租户级、工具级的多层次限流"""
def __init__(self, redis_url: str):
self.redis_url = redis_url
self._local_burst_cache: Dict[str, Tuple[int, float]] = {}
self._rules: Dict[str, RateLimitRule] = {
"default": RateLimitRule(100, 1000, 20),
"company_search": RateLimitRule(500, 5000, 50),
"risk_scan": RateLimitRule(20, 200, 5),
"legal_case_query": RateLimitRule(100, 1000, 20),
}
async def check(
self,
tenant_id: str,
tool_name: str,
trace_id: str
) -> Tuple[bool, dict]:
"""
检查请求是否允许通过
返回: (是否允许, 限流元信息)
"""
rule = self._rules.get(tool_name, self._rules["default"])
now = time.time()
# 突发限流（本地内存，毫秒级）
burst_key = f"{tenant_id}:{tool_name}"
if burst_key in self._local_burst_cache:
count, window_start = self._local_burst_cache[burst_key]
window_duration = now - window_start
if window_duration < 1.0: # 1秒窗口
if count >= rule.burst_size:
return False, {
"reason": "burst_limit",
"retry_after_ms": int(1000 - window_duration * 1000)
}
self._local_burst_cache[burst_key] = (count + 1, window_start)
else:
self._local_burst_cache[burst_key] = (1, now)
else:
self._local_burst_cache[burst_key] = (1, now)
# 分钟级限流（Redis）
minute_key = f"rl:minute:{tenant_id}:{tool_name}"
minute_count = await self.redis.get(minute_key)
if minute_count and int(minute_count) >= rule.requests_per_minute:
ttl = await self.redis.ttl(minute_key)
return False, {
"reason": "minute_limit",
"retry_after_ms": ttl * 1000
}
# 小时级限流（Redis）
hour_key = f"rl:hour:{tenant_id}:{tool_name}"
hour_count = await self.redis.get(hour_key)
if hour_count and int(hour_count) >= rule.requests_per_hour:
ttl = await self.redis.ttl(hour_key)
return False, {
"reason": "hour_limit",
"retry_after_ms": ttl * 1000
}
# 记录请求
pipe = self.redis.pipeline()
pipe.incr(minute_key)
pipe.expire(minute_key, 60)
pipe.incr(hour_key)
pipe.expire(hour_key, 3600)
await pipe.execute()
return True, {"allowed": True}
def get_headers(self, metadata: dict) -> dict:
"""生成返回给客户端的限流头"""
headers = {}
if "retry_after_ms" in metadata:
headers["Retry-After"] = str(metadata["retry_after_ms"] // 1000)
headers["X-RateLimit-Retry-After-Ms"] = str(metadata["retry_after_ms"])
return headers
A.4 证据元信息生成器
# src/utils/evidence.py
import hashlib
import uuid
from datetime import datetime, timezone
from typing import List, Optional
from dataclasses import dataclass, asdict
@dataclass
class Evidence:
"""MCP v0.28 证据元信息：每个工具调用结果的核心组成部分"""
evidence_id: str
trace_id: str
data_source: str
query_timestamp: str
data_timepoint: str
confidence: float
limitations: List[str]
fields_used: List[str]
raw_response_hash: str
server_instance: str
protocol_version: str = "0.28.0"
cache_hit: bool = False
def to_meta(self) -> dict:
"""转换为返回给客户端的 _meta 字段"""
return {
"evidence_id": self.evidence_id,
"trace_id": self.trace_id,
"data_source": self.data_source,
"query_timestamp": self.query_timestamp,
"data_timepoint": self.data_timepoint,
"confidence": self.confidence,
"limitations": self.limitations,
"fields_used": self.fields_used,
"raw_response_hash": self.raw_response_hash,
"server_instance": self.server_instance,
"cache_hit": self.cache_hit,
"protocol_version": self.protocol_version
}
class EvidenceGenerator:
def __init__(self, server_instance: str, redis_url: str):
self.server_instance = server_instance
self.redis = None # 懒加载
@staticmethod
def hash_response(data: dict) -> str:
"""对响应数据生成哈希，用于审计"""
import json
normalized = json.dumps(data, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(normalized.encode()).hexdigest()[:16]
async def generate(
self,
trace_id: str,
tool_name: str,
data: dict,
data_source: str,
data_timepoint: str,
confidence: float,
limitations: List[str],
fields_used: List[str],
cache_hit: bool = False
) -> Evidence:
"""生成证据元信息"""
return Evidence(
evidence_id=f"ev-{uuid.uuid4().hex[:12]}",
trace_id=trace_id,
data_source=data_source,
query_timestamp=datetime.now(timezone.utc).isoformat(),
data_timepoint=data_timepoint,
confidence=confidence,
limitations=limitations,
fields_used=fields_used,
raw_response_hash=self.hash_response(data),
server_instance=self.server_instance,
cache_hit=cache_hit
)
A.5 主服务器入口
# src/server.py
import asyncio
import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import structlog
from opentelemetry import trace
from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server
from src.governance.rate_limiter import TieredRateLimiter
from src.utils.evidence import EvidenceGenerator
logger = structlog.get_logger()
tracer = trace.get_tracer(__name__)
app = FastAPI(title="Enterprise MCP Server v0.28")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# 全局组件（通过 lifespan 管理生命周期）
rate_limiter: TieredRateLimiter = None
evidence_gen: EvidenceGenerator = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global rate_limiter, evidence_gen
import aioredis
redis = await aioredis.from_url("redis://localhost:6379")
rate_limiter = TieredRateLimiter("redis://localhost:6379")
evidence_gen = EvidenceGenerator(
server_instance=f"server-{uuid.uuid4().hex[:8]}",
redis_url="redis://localhost:6379"
)
logger.info("server.started", instance=evidence_gen.server_instance)
yield
await redis.close()
logger.info("server.stopped")
app.router.lifespan_context = lifespan
# ========== MCP v0.28 HTTP 传输 ==========
class ToolCallRequest(BaseModel):
name: str
arguments: dict
trace_id: Optional[str] = None
tenant_id: Optional[str] = None
meta: Optional[dict] = None
@app.post("/tools/call")
async def call_tool(req: ToolCallRequest, request: Request) -> Response:
trace_id = req.trace_id or uuid.uuid4().hex
tenant_id = req.tenant_id or "anonymous"
with tracer.start_as_current_span(f"tool.{req.name}") as span:
span.set_attribute("trace_id", trace_id)
span.set_attribute("tenant_id", tenant_id)
span.set_attribute("tool_name", req.name)
# Step 1: 限流检查
allowed, limit_meta = await rate_limiter.check(
tenant_id, req.name, trace_id
)
if not allowed:
headers = rate_limiter.get_headers(limit_meta)
return Response(
status_code=429,
content='{"error": "rate_limit_exceeded"}',
headers={**headers, "Content-Type": "application/json"}
)
# Step 2: 执行工具
try:
result = await execute_tool(req.name, req.arguments)
# Step 3: 生成证据元信息
evidence = await evidence_gen.generate(
trace_id=trace_id,
tool_name=req.name,
data=result,
data_source="business_database",
data_timepoint="current",
confidence=0.95,
limitations=["仅覆盖中国大陆企业数据"],
fields_used=list(result.keys()),
cache_hit=False
)
response_body = {
"result": result,
"_meta": evidence.to_meta()
}
span.set_attribute("success", True)
return Response(
content=json.dumps(response_body),
headers={
"Content-Type": "application/json",
"X-Trace-Id": trace_id,
"X-Evidence-Id": evidence.evidence_id
}
)
except Exception as e:
span.record_exception(e)
logger.error(
"tool.execution_failed",
tool=req.name,
trace_id=trace_id,
error=str(e)
)
raise HTTPException(status_code=500, detail=str(e))
@app.get("/tools/list")
async def list_tools():
"""返回带完整 v0.28 annotations 的工具清单"""
return {
"tools": [
{
"name": "company_search",
"description": "模糊搜索企业名称，返回匹配的企业列表",
"inputSchema": CompanySearchInput.model_json_schema(),
"outputSchema": CompanySearchOutput.model_json_schema(),
"annotations": {
"dataScope": "current",
"dataFreshness": "realtime",
"capabilityType": "search",
"requiresAnchor": False,
"prerequisites": [],
"mutuallyExclusiveWith": [],
"cacheTtlMs": 300000, # 5分钟缓存
"costUnits": 2,
"rateLimitTier": "default"
}
},
{
"name": "risk_scan",
"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 dispatch[name](arguments)
async def do_company_search(args: dict) -> dict:
# 实际实现：调用数据库或上游 API
return {"companies": [], "total": 0, "search_id": uuid.uuid4().hex}
A.6 Docker 部署配置
# 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）原创，版权所有。