第7讲：实战——API 网关 MCP Server

第5讲和第6讲我们分别实现了数据库和文件系统 MCP Server。这两个场景有一个共同点：它们操作的是“本地资源”。但在真实世界中，Agent 还需要调用外部 API——CRM 系统、企业微信、GitHub、Jira、内部微服务……

这一讲，我们要实现一个API 网关 MCP Server，把任意 HTTP API 包装成 MCP 工具。学完这一讲，你的 Agent 就能调用任何外部服务了。

一、需求分析

1.1 场景

• 查询天气：调用第三方天气 API

• 发送消息：调用企业微信/钉钉 Webhook

• 管理工单：调用 Jira API 创建/查询工单

• 代码管理：调用 GitHub API 创建 Issue、查看 PR

1.2 核心挑战

二、架构设计

用户 → Agent → MCP Client → API Gateway MCP Server → 外部 API ├── 认证管理器 ├── 请求构造器 ├── 响应解析器 ├── 限流保护 └── 错误处理器

三、实现 API 网关 MCP Server

3.1 API 配置定义

首先定义一个灵活的 API 配置模型，用来描述每个 API 接口：

# api_config.py from dataclasses import dataclass, field from typing import Optional, Any from enum import Enum class AuthType(Enum): NONE = "none" API_KEY = "api_key" # Header: X-API-Key BEARER = "bearer" # Header: Authorization: Bearer xxx BASIC = "basic" # Basic Auth CUSTOM = "custom" # 自定义认证 class ParamLocation(Enum): QUERY = "query" # URL 参数 HEADER = "header" # HTTP 头 BODY = "body" # 请求体 (JSON) PATH = "path" # URL 路径 @dataclass class APIParam: """API 参数定义""" name: str location: ParamLocation type: str = "string" # string, number, boolean, object description: str = "" required: bool = True default: Any = None @dataclass class APIDefinition: """API 接口定义""" name: str # 工具名称 description: str # 工具描述 method: str = "GET" # HTTP 方法 url_template: str = "" # URL 模板，如 https://api.github.com/repos/{owner}/{repo} auth_type: AuthType = AuthType.NONE params: list[APIParam] = field(default_factory=list) headers: dict = field(default_factory=dict) timeout: int = 30 success_codes: list[int] = field(default_factory=lambda: [200, 201, 204])

3.2 API 注册中心

# api_registry.py from typing import Dict, List from api_config import APIDefinition, APIParam, ParamLocation, AuthType class APIRegistry: """API 注册中心：管理所有可用的 API 定义""" def __init__(self): self._apis: Dict[str, APIDefinition] = {} def register(self, api: APIDefinition): """注册一个 API""" self._apis[api.name] = api def get(self, name: str) -> APIDefinition: """获取 API 定义""" if name not in self._apis: raise KeyError(f"API '{name}' 未注册") return self._apis[name] def list_all(self) -> List[APIDefinition]: """列出所有 API""" return list(self._apis.values()) def register_from_dict(self, configs: list[dict]): """从字典列表批量注册""" for cfg in configs: params = [ APIParam(**p) if isinstance(p, dict) else p for p in cfg.get("params", []) ] api = APIDefinition( name=cfg["name"], description=cfg.get("description", ""), method=cfg.get("method", "GET"), url_template=cfg["url_template"], auth_type=AuthType(cfg.get("auth_type", "none")), params=params, headers=cfg.get("headers", {}), timeout=cfg.get("timeout", 30) ) self.register(api) # 内置一些常用的 API 定义 def create_default_registry() -> APIRegistry: """创建默认的 API 注册表""" registry = APIRegistry() # 1. 天气查询（无需认证） registry.register(APIDefinition( name="get_weather", description="查询指定城市的实时天气信息", method="GET", url_template="https://api.openweathermap.org/data/2.5/weather", params=[ APIParam(name="city", location=ParamLocation.QUERY, description="城市名称，如 Beijing、Shanghai"), APIParam(name="units", location=ParamLocation.QUERY, description="温度单位: metric(摄氏度) 或 imperial(华氏度)", required=False, default="metric"), ] )) # 2. GitHub API 示例 registry.register(APIDefinition( name="github_get_repo", description="获取 GitHub 仓库信息", method="GET", url_template="https://api.github.com/repos/{owner}/{repo}", params=[ APIParam(name="owner", location=ParamLocation.PATH, description="仓库所有者"), APIParam(name="repo", location=ParamLocation.PATH, description="仓库名称"), ], auth_type=AuthType.BEARER )) # 3. 企业微信机器人消息 registry.register(APIDefinition( name="send_wechat_message", description="通过企业微信群机器人发送消息", method="POST", url_template="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={webhook_key}", params=[ APIParam(name="webhook_key", location=ParamLocation.PATH, description="群机器人的 Webhook Key"), APIParam(name="content", location=ParamLocation.BODY, description="消息内容"), APIParam(name="msg_type", location=ParamLocation.BODY, description="消息类型: text(文本) 或 markdown", required=False, default="text"), ], timeout=10 )) return registry

3.3 认证管理器

# auth_manager.py import base64 from typing import Optional from api_config import AuthType class AuthManager: """认证管理器：管理 API Key、Token 等凭证""" def __init__(self): # 存储不同服务的凭证 self._credentials: dict[str, dict] = {} def set_credential(self, service: str, credential: dict): """设置某个服务的凭证""" self._credentials[service] = credential def get_auth_headers(self, auth_type: AuthType, service: str = "") -> dict: """根据认证类型生成 HTTP Headers""" cred = self._credentials.get(service, {}) if auth_type == AuthType.NONE: return {} elif auth_type == AuthType.API_KEY: key = cred.get("api_key", "") header_name = cred.get("header_name", "X-API-Key") return {header_name: key} elif auth_type == AuthType.BEARER: token = cred.get("token", "") return {"Authorization": f"Bearer {token}"} elif auth_type == AuthType.BASIC: username = cred.get("username", "") password = cred.get("password", "") encoded = base64.b64encode(f"{username}:{password}".encode()).decode() return {"Authorization": f"Basic {encoded}"} return {} def get_credential_names(self) -> list[str]: """获取所有已设置的凭证名称""" return list(self._credentials.keys())

3.4 请求执行器

# request_executor.py import json import httpx import asyncio from typing import Optional, Any from datetime import datetime from api_config import APIDefinition, ParamLocation, AuthType from auth_manager import AuthManager class RequestExecutor: """HTTP 请求执行器""" def __init__(self, auth_manager: AuthManager, timeout: int = 30): self.auth = auth_manager self.default_timeout = timeout self._client = httpx.Client(timeout=timeout) def execute(self, api_def: APIDefinition, params: dict) -> dict: """ 执行 API 请求 返回： { "success": bool, "status_code": int, "data": dict or str, "error": str or None, "duration_ms": float } """ start = datetime.now() try: # 1. 构建 URL url = self._build_url(api_def, params) # 2. 分离参数 query_params, headers, body = self._separate_params(api_def, params) # 3. 添加认证信息 auth_headers = self.auth.get_auth_headers( api_def.auth_type, api_def.name.split("_")[0] # 用 API 前缀作为 service 名 ) headers.update(auth_headers) headers.update(api_def.headers) # 4. 发送请求 response = self._send_request( api_def.method, url, query_params, headers, body ) duration = (datetime.now() - start).total_seconds() * 1000 # 5. 处理响应 if response.status_code in api_def.success_codes: try: data = response.json() except: data = response.text return { "success": True, "status_code": response.status_code, "data": data, "error": None, "duration_ms": round(duration, 1) } else: return { "success": False, "status_code": response.status_code, "data": None, "error": f"HTTP {response.status_code}: {response.text[:200]}", "duration_ms": round(duration, 1) } except httpx.TimeoutException: duration = (datetime.now() - start).total_seconds() * 1000 return { "success": False, "status_code": 0, "data": None, "error": f"请求超时 ({api_def.timeout}s)", "duration_ms": round(duration, 1) } except Exception as e: duration = (datetime.now() - start).total_seconds() * 1000 return { "success": False, "status_code": 0, "data": None, "error": str(e), "duration_ms": round(duration, 1) } def _build_url(self, api_def: APIDefinition, params: dict) -> str: """构建最终 URL（替换路径参数）""" url = api_def.url_template for param in api_def.params: if param.location == ParamLocation.PATH: value = params.get(param.name, "") url = url.replace(f"{{{param.name}}}", str(value)) return url def _separate_params(self, api_def: APIDefinition, params: dict) -> tuple: """将参数按位置分离""" query_params = {} headers = {} body = {} for param in api_def.params: value = params.get(param.name, param.default) if value is None: continue if param.location == ParamLocation.QUERY: query_params[param.name] = value elif param.location == ParamLocation.HEADER: headers[param.name] = value elif param.location == ParamLocation.BODY: body[param.name] = value return query_params, headers, body def _send_request(self, method: str, url: str, params: dict, headers: dict, body: dict): """发送 HTTP 请求""" method = method.upper() if method == "GET": return self._client.get(url, params=params, headers=headers) elif method == "POST": return self._client.post(url, params=params, headers=headers, json=body) elif method == "PUT": return self._client.put(url, params=params, headers=headers, json=body) elif method == "DELETE": return self._client.delete(url, params=params, headers=headers) elif method == "PATCH": return self._client.patch(url, params=params, headers=headers, json=body) else: raise ValueError(f"不支持的 HTTP 方法: {method}") def close(self): self._client.close()

3.5 主 Server 文件

# api_mcp_server.py import json import os from mcp.server import Server from mcp.server.stdio import stdio_server import mcp.types as types from api_config import APIDefinition, APIParam, ParamLocation, AuthType from api_registry import APIRegistry, create_default_registry from auth_manager import AuthManager from request_executor import RequestExecutor # 初始化组件 registry = create_default_registry() auth_manager = AuthManager() executor = RequestExecutor(auth_manager) server = Server("api-gateway-server") # 从环境变量加载凭证 def load_credentials_from_env(): """从环境变量加载 API 凭证""" # 天气 API if weather_key := os.environ.get("OPENWEATHER_API_KEY"): auth_manager.set_credential("get_weather", { "api_key": weather_key, "header_name": "appid" # OpenWeatherMap 用 appid 参数名 }) # GitHub Token if github_token := os.environ.get("GITHUB_TOKEN"): auth_manager.set_credential("github", {"token": github_token}) load_credentials_from_env() @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """动态生成工具列表""" tools = [] for api_def in registry.list_all(): # 构建参数 Schema properties = {} required = [] for param in api_def.params: param_type = param.type if param_type == "object": param_type = "string" # 简化处理 prop = { "type": param_type, "description": param.description } if param.default is not None: prop["default"] = param.default properties[param.name] = prop if param.required: required.append(param.name) tools.append(types.Tool( name=api_def.name, description=f"{api_def.description}\n\nHTTP {api_def.method} {api_def.url_template}", inputSchema={ "type": "object", "properties": properties, "required": required } )) return tools @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]: try: # 查找 API 定义 api_def = registry.get(name) # 执行请求 result = executor.execute(api_def, arguments) if result["success"]: # 格式化成功响应 formatted = json.dumps(result["data"], ensure_ascii=False, indent=2) meta = f"\n\n⏱ {result['duration_ms']}ms | HTTP {result['status_code']}" return [types.TextContent(type="text", text=formatted + meta)] else: # 格式化错误响应 error_msg = f"请求失败: {result['error']}" if result["status_code"]: error_msg += f" (HTTP {result['status_code']})" return [types.TextContent(type="text", text=error_msg)] except KeyError: return [types.TextContent(type="text", text=f"未知 API: {name}")] except Exception as e: return [types.TextContent(type="text", text=f"执行异常: {str(e)}")] async def main(): async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options() ) if __name__ == "__main__": asyncio.run(main())

四、测试 API 网关 MCP Server

4.1 安装依赖

pip install httpx

4.2 设置环境变量（可选）

# 如果有 API Key 可以设置 export OPENWEATHER_API_KEY=your_api_key_here export GITHUB_TOKEN=your_github_token_here

4.3 测试 Client

# test_api_client.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test(): server_params = StdioServerParameters( command="python", args=["api_mcp_server.py"] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 1. 列出所有可用 API print("=== 可用 API ===") tools = await session.list_tools() for tool in tools.tools: print(f" • {tool.name}: {tool.description.split(chr(10))[0]}") # 2. 测试天气查询（如果设置了 API Key） print("\n=== 查询天气 ===") result = await session.call_tool("get_weather", { "city": "Beijing", "units": "metric" }) print(result.content[0].text[:300]) # 3. 测试 GitHub API（如果设置了 Token） print("\n=== 查询 GitHub 仓库 ===") result = await session.call_tool("github_get_repo", { "owner": "torvalds", "repo": "linux" }) print(result.content[0].text[:300]) if __name__ == "__main__": asyncio.run(test())

五、动态注册 API（扩展）

为了让 API 网关更灵活，我们可以支持从配置文件动态加载 API 定义：

5.1 API 配置文件

# apis.yaml apis: - name: "get_joke" description: "获取一条随机笑话" method: GET url_template: "https://v2.jokeapi.dev/joke/Any" auth_type: none params: [] - name: "get_quote" description: "获取一条随机名言" method: GET url_template: "https://api.quotable.io/random" auth_type: none params: [] - name: "create_jira_issue" description: "在 Jira 中创建一条 Issue" method: POST url_template: "https://{domain}.atlassian.net/rest/api/3/issue" auth_type: basic params: - name: domain location: path type: string description: "Jira 域名（不含 .atlassian.net）" - name: project_key location: body type: string description: "项目 Key" - name: summary location: body type: string description: "Issue 标题" - name: description location: body type: string description: "Issue 描述" required: false

5.2 动态加载器

# dynamic_loader.py import yaml from api_registry import APIRegistry from api_config import APIDefinition, APIParam, ParamLocation, AuthType def load_apis_from_yaml(filepath: str) -> APIRegistry: """从 YAML 文件加载 API 定义""" registry = APIRegistry() with open(filepath, "r", encoding="utf-8") as f: data = yaml.safe_load(f) for cfg in data.get("apis", []): params = [] for p in cfg.get("params", []): params.append(APIParam( name=p["name"], location=ParamLocation(p.get("location", "query")), type=p.get("type", "string"), description=p.get("description", ""), required=p.get("required", True), default=p.get("default") )) api = APIDefinition( name=cfg["name"], description=cfg.get("description", ""), method=cfg.get("method", "GET"), url_template=cfg["url_template"], auth_type=AuthType(cfg.get("auth_type", "none")), params=params, timeout=cfg.get("timeout", 30) ) registry.register(api) return registry

六、安全最佳实践

6.1 凭证管理

# 不要硬编码凭证！ # 错误做法： auth_manager.set_credential("github", {"token": "ghp_xxxxxxxxxxxx"}) # 正确做法：从环境变量读取 import os github_token = os.environ.get("GITHUB_TOKEN") if github_token: auth_manager.set_credential("github", {"token": github_token}) # 更安全的做法：使用 secrets 管理服务 # 如 HashiCorp Vault、AWS Secrets Manager

6.2 请求白名单

ALLOWED_DOMAINS = [ "api.openweathermap.org", "api.github.com", "qyapi.weixin.qq.com", # 公司内部 API "api.internal.company.com", ] def validate_url(url: str) -> bool: """验证请求 URL 是否在白名单中""" from urllib.parse import urlparse parsed = urlparse(url) return any(parsed.netloc.endswith(domain) for domain in ALLOWED_DOMAINS)

6.3 敏感信息过滤

SENSITIVE_FIELDS = ["password", "token", "secret", "key", "authorization"] def filter_sensitive_data(data: dict) -> dict: """过滤响应中的敏感信息""" filtered = {} for key, value in data.items(): if any(s in key.lower() for s in SENSITIVE_FIELDS): filtered[key] = "***" elif isinstance(value, dict): filtered[key] = filter_sensitive_data(value) else: filtered[key] = value return filtered

七、课后作业

1. 添加更多内置 API：在create_default_registry()中添加以下 API：

   • get_random_cat_image：获取随机猫咪图片

   • get_exchange_rate：获取实时汇率

   • shorten_url：URL 缩短服务

2. 实现请求缓存：对于 GET 请求，在 5 分钟内返回相同参数的缓存结果，减少外部 API 调用。

3. 挑战题：实现一个“API 组合工具”——允许 Agent 在一次调用中串联多个 API（如先查天气，再把结果发送到企业微信）。

八、总结

这一讲我们完成了：

• API 网关 MCP Server：将任意 HTTP API 包装成 MCP 工具

• 灵活的配置模型：支持 GET/POST/PUT/DELETE，多种参数位置和认证方式

• 动态注册机制：从代码或配置文件加载 API 定义

• 完善的错误处理：超时、HTTP 错误码、异常捕获

• 安全防护：凭证管理、域名白名单、敏感信息过滤

你现在拥有了一个通用的 API 网关，可以对接任何外部服务。结合前两讲的数据库和文件系统 Server，你的 Agent 已经能连接三类最常见的系统了。

下一讲，我们将进入进阶篇——把 MCP Server 和 Agent 深度集成，让 Agent 能自主发现和使用 MCP 工具，并实现多 Server 编排。

🧰 开发之余，处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求，我常用一个纯前端本地工具箱：zz365.top（子页 PDF 大师：PDF 大师 - zz365工具箱）。所有计算在浏览器完成，文件不上传服务器，关页即清。免费、无登录、无广告，适合开发者当常驻标签页。