第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、查看 PR1.2 核心挑战挑战说明认证管理​不同 API 有不同的认证方式API Key、OAuth、Basic Auth参数映射​HTTP 参数Header、Query、Body需要映射到 MCP 参数错误处理​HTTP 错误码需要转换为友好的错误信息限流保护​防止 Agent 过度调用外部 API 导致被封二、架构设计用户 → Agent → MCP Client → API Gateway MCP Server → 外部 API ├── 认证管理器 ├── 请求构造器 ├── 响应解析器 ├── 限流保护 └── 错误处理器三、实现 API 网关 MCP Server3.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_factorylist) headers: dict field(default_factorydict) timeout: int 30 success_codes: list[int] field(default_factorylambda: [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(fAPI {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( namecfg[name], descriptioncfg.get(description, ), methodcfg.get(method, GET), url_templatecfg[url_template], auth_typeAuthType(cfg.get(auth_type, none)), paramsparams, headerscfg.get(headers, {}), timeoutcfg.get(timeout, 30) ) self.register(api) # 内置一些常用的 API 定义 def create_default_registry() - APIRegistry: 创建默认的 API 注册表 registry APIRegistry() # 1. 天气查询无需认证 registry.register(APIDefinition( nameget_weather, description查询指定城市的实时天气信息, methodGET, url_templatehttps://api.openweathermap.org/data/2.5/weather, params[ APIParam(namecity, locationParamLocation.QUERY, description城市名称如 Beijing、Shanghai), APIParam(nameunits, locationParamLocation.QUERY, description温度单位: metric(摄氏度) 或 imperial(华氏度), requiredFalse, defaultmetric), ] )) # 2. GitHub API 示例 registry.register(APIDefinition( namegithub_get_repo, description获取 GitHub 仓库信息, methodGET, url_templatehttps://api.github.com/repos/{owner}/{repo}, params[ APIParam(nameowner, locationParamLocation.PATH, description仓库所有者), APIParam(namerepo, locationParamLocation.PATH, description仓库名称), ], auth_typeAuthType.BEARER )) # 3. 企业微信机器人消息 registry.register(APIDefinition( namesend_wechat_message, description通过企业微信群机器人发送消息, methodPOST, url_templatehttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?key{webhook_key}, params[ APIParam(namewebhook_key, locationParamLocation.PATH, description群机器人的 Webhook Key), APIParam(namecontent, locationParamLocation.BODY, description消息内容), APIParam(namemsg_type, locationParamLocation.BODY, description消息类型: text(文本) 或 markdown, requiredFalse, defaulttext), ], timeout10 )) return registry3.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: fBearer {token}} elif auth_type AuthType.BASIC: username cred.get(username, ) password cred.get(password, ) encoded base64.b64encode(f{username}:{password}.encode()).decode() return {Authorization: fBasic {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(timeouttimeout) 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: fHTTP {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, paramsparams, headersheaders) elif method POST: return self._client.post(url, paramsparams, headersheaders, jsonbody) elif method PUT: return self._client.put(url, paramsparams, headersheaders, jsonbody) elif method DELETE: return self._client.delete(url, paramsparams, headersheaders) elif method PATCH: return self._client.patch(url, paramsparams, headersheaders, jsonbody) 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( nameapi_def.name, descriptionf{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_asciiFalse, indent2) meta f\n\n⏱ {result[duration_ms]}ms | HTTP {result[status_code]} return [types.TextContent(typetext, textformatted meta)] else: # 格式化错误响应 error_msg f请求失败: {result[error]} if result[status_code]: error_msg f (HTTP {result[status_code]}) return [types.TextContent(typetext, texterror_msg)] except KeyError: return [types.TextContent(typetext, textf未知 API: {name})] except Exception as e: return [types.TextContent(typetext, textf执行异常: {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 Server4.1 安装依赖pip install httpx4.2 设置环境变量可选# 如果有 API Key 可以设置 export OPENWEATHER_API_KEYyour_api_key_here export GITHUB_TOKENyour_github_token_here4.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( commandpython, 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: false5.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, encodingutf-8) as f: data yaml.safe_load(f) for cfg in data.get(apis, []): params [] for p in cfg.get(params, []): params.append(APIParam( namep[name], locationParamLocation(p.get(location, query)), typep.get(type, string), descriptionp.get(description, ), requiredp.get(required, True), defaultp.get(default) )) api APIDefinition( namecfg[name], descriptioncfg.get(description, ), methodcfg.get(method, GET), url_templatecfg[url_template], auth_typeAuthType(cfg.get(auth_type, none)), paramsparams, timeoutcfg.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 Manager6.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七、课后作业添加更多内置 API在create_default_registry()中添加以下 APIget_random_cat_image获取随机猫咪图片get_exchange_rate获取实时汇率shorten_urlURL 缩短服务实现请求缓存对于 GET 请求在 5 分钟内返回相同参数的缓存结果减少外部 API 调用。挑战题实现一个“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工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。