架构愿景:从数据孤岛到智能协同
2.1 核心架构理念
FastMCP服务作为业务系统与AI模型之间的中间层,实现了三大核心价值:
┌─────────────────────────────────────────────────────────────┐
│ AI 大模型(如GPT、LLM) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent / Function Calling │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ FastMCP Service Layer │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ • API Key 认证 • 接口路由管理 │ │ │
│ │ │ • 参数标准化 • 安全策略执行 │ │ │
│ │ │ • 请求日志审计 • 流量控制限流 │ │ │
│ │ └─────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 业务系统层(Node.js服务) │ │
│ │ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ /api/alarms│ │ /api/assets │ • 告警管理 │ │
│ │ │ (告警查询) │ │ (资产管理) │ • 资产管理 │ │
│ │ └─────────────┘ └─────────────┘ • 配置管理 │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
2.2 设计原则
| 原则 | 实现策略 | 价值体现 |
|---|---|---|
| 松耦合 | 通过HTTP接口抽象业务能力 | 业务系统独立演进,不依赖AI技术栈 |
| 安全性 | API Key认证、请求审计 | 确保敏感业务数据安全可控 |
| 可扩展 | 动态接口注册机制 | 快速接入新业务能力 |
| 标准化 | MCP协议规范 | AI模型可统一调用各类业务系统 |
三、技术实现:FastMCP服务核心机制
3.1 服务初始化与接口注册
# FastMCP服务初始化 mcp = FastMCP() registered_tools: List[str] = [] def register_all_tools() -> None: """动态注册所有业务接口作为MCP工具""" for interface in store.list_interfaces(): registered_tools.append(create_tool_for_interface(interface))技术要点:
- 通过
interface_store持久化管理接口元数据 - 支持接口的CRUD操作,无需重启服务即可更新能力列表
- 接口定义包含URL、HTTP方法、参数描述等完整信息
3.2 动态工具生成机制
FastMCP的核心创新在于能够根据接口元数据自动生成可被AI调用的工具函数:
def create_tool_for_interface(interface: Dict[str, Any]) -> str: """根据接口定义动态生成MCP工具""" interface_id = interface["id"] interface_path = interface["url"] interface_method = interface["method"] interface_parameters = interface.get("parameters", {}) # 提取URL路径参数(如 /api/asset/:id) path_params = [] import re path_param_pattern = re.compile(r':([a-zA-Z0-9_]+)') matches = path_param_pattern.findall(interface_path) for match in matches: path_params.append(match) # 构建工具函数签名 # 支持路径参数替换、请求体参数组装、动态代码执行 ...核心技术亮点:
- 参数标准化:自动将URL路径参数(如
:id)转换为函数参数 - 动态代码生成:使用
exec()动态创建符合MCP规范的工具函数 - 智能参数映射:支持复杂参数结构的自动转换
3.3 API Key认证体系
安全性是企业级应用的基石,FastMCP服务实现了灵活的API Key认证机制:
# 支持多种API Key配置方式 DEFAULT_API_KEYS = ["api-key-001", "api-key-002"] MCP_API_KEY = os.getenv("MCP_API_KEY", None) env_api_keys = os.getenv("MCP_API_KEYS", "") MCP_API_KEYS = env_api_keys.split(",") if env_api_keys else DEFAULT_API_KEYS def api_key_auth_middleware(request): """API Key认证中间件""" if not MCP_API_KEY and not MCP_API_KEYS: return request # 支持两种认证方式 auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): provided_key = auth_header[7:] else: provided_key = request.headers.get("X-API-Key", "") # 多Key验证机制 valid_keys = [] if MCP_API_KEY: valid_keys.append(MCP_API_KEY) if MCP_API_KEYS: valid_keys.extend([k.strip() for k in MCP_API_KEYS if k.strip()]) if not provided_key or provided_key not in valid_keys: raise HTTPException(status_code=401, detail="Unauthorized") return request安全特性:
- 灵活配置:支持环境变量配置或代码内置默认Key
- 多Key支持:允许同时配置多个有效API Key
- 双Header支持:兼容
Authorization: Bearer和X-API-Key两种方式 - 渐进式安全:未配置Key时自动跳过认证,便于开发调试
3.4 HTTP请求代理机制
作为业务系统的代理层,FastMCP负责将标准化的工具调用转换为实际的HTTP请求:
def make_request(url: str, method: str = "GET", data: Dict[str, Any] = None) -> Dict[str, Any]: """统一的HTTP请求封装""" try: if data is not None and method.upper() != "GET": data_json = json.dumps(data).encode("utf-8") headers = {"Content-Type": "application/json"} req = urllib.request.Request(url, data=data_json, headers=headers, method=method) else: req = urllib.request.Request(url, method=method) with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT_SECONDS) as response: response_data = response.read().decode("utf-8") return json.loads(response_data) except urllib.error.HTTPError as e: # 上游服务错误处理 ...可靠性保障:
- 超时控制:防止慢请求导致服务阻塞
- 错误透传:保留上游服务的错误信息便于排查
- 协议适配:自动处理GET/POST等不同HTTP方法
四、应用场景:解锁业务数据价值
4.1 智能告警分析
AI模型通过调用api_1工具查询告警数据: