🧠 图解记忆: Gateway 管入口和策略,Registry 管能力和版本,Server 隔离执行真实业务。
💡 答案要点
企业MCP架构:
┌─────────────────────────────────────────────────────────────────┐
│ 企业级 MCP 部署架构 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ MCP Gateway(网关层) │ │
│ │ - 统一入口,认证鉴权 │ │
│ │ - 请求路由(分发到对应MCP Server) │ │
│ │ - 限流熔断 │ │
│ │ - 审计日志 │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌───────────────────┼───────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ MCP Server │ │ MCP Server │ │ MCP Server │ │
│ │ 文件系统 │ │ 数据库 │ │ GitHub │ │
│ │ (本地隔离) │ │ (内网隔离) │ │ (企业账号) │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ 工具注册中心(Registry) │ │
│ │ - 工具元数据存储 │ │
│ │ - 版本管理 │ │
│ │ - 可用性监控 │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘MCP Gateway实现:
展开 Python 代码示例(53 行)
python
# mcp_gateway.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer
import httpx
import asyncio
app = FastAPI(title="企业MCP Gateway")
# 路由配置
ROUTES = {
"filesystem": "http://localhost:8001",
"database": "http://localhost:8002",
"github": "http://localhost:8003",
}
# OAuth认证
oauth = OAuth2PasswordBearer(tokenUrl="token")
@app.post("/mcp/{server_name}/tools/call")
async def route_tool_call(
server_name: str,
request: dict,
token: str = Depends(oauth)
):
# 1. 验证Token
user = await verify_token(token)
await check_scope(user, "tool:call")
# 2. 路由到对应Server
if server_name not in ROUTES:
raise HTTPException(404, "Server not found")
# 3. 限流检查
await rate_limit.check(user.id, server_name)
# 4. 转发请求
async with httpx.AsyncClient() as client:
response = await client.post(
f"{ROUTES[server_name]}/tools/call",
json=request,
headers={"Authorization": f"Bearer {token}"},
timeout=30
)
# 5. 审计日志
await audit_log.record(
user=user.id,
server=server_name,
tool=request.get("name"),
timestamp=datetime.utcnow()
)
return response.json()私有MCP Server部署:
yaml
# docker-compose.yml(文件Server)
services:
mcp-filesystem:
image: python:3.11-slim
command: python -m mcp_filesystem_server ./data --port 8001
volumes:
- ./data:/data:ro # 只读挂载,防止写入敏感目录
environment:
- ALLOWED_DIRECTORIES=/data # 限制只能访问/data
networks:
- mcp-internal
mcp-gateway:
image: nginx + fastapi
ports:
- "8080:8080"
depends_on:
- mcp-filesystem
networks:
- mcp-internal
- mcp-external
networks:
mcp-internal:
internal: true # 外网无法直接访问内部Server
mcp-external:企业MCP Registry(工具注册中心):
展开 Python 代码示例(32 行)
python
# MCP Registry API
@app.get("/registry/tools")
async def list_tools(
category: str = None,
tags: list[str] = None,
page: int = 1,
page_size: int = 20
):
"""列出所有注册的工具"""
query = {"status": "active"}
if category:
query["category"] = category
if tags:
query["tags"] = {"$all": tags}
tools = await registry.find(query, skip=(page-1)*page_size, limit=page_size)
total = await registry.count(query)
return {
"items": [format_tool(tool) for tool in tools],
"total": total,
"page": page,
"page_size": page_size
}
@app.post("/registry/tools/{tool_id}/versions")
async def publish_version(tool_id: str, version: ToolVersion):
"""发布新版本"""
# 版本兼容性检查
# 灰度发布
# 通知使用方
return {"version": version.id, "status": "published"}MCP监控指标:
| 指标 | 说明 | 告警阈值 |
|---|---|---|
| 工具调用成功率 | 成功/总调用 | < 99% |
| 工具调用延迟P99 | 第99百分位延迟 | > 2s |
| 工具错误率 | 各类错误分布 | > 1% |
| Token消耗 | 各Server资源使用 | 按配额 |
| Scope违规 | 越权调用尝试 | > 0 |
面试话术:
"企业MCP规模化部署的核心是Gateway+Registry架构。Gateway做统一入口(认证、路由、限流、审计),Registry做工具注册(版本管理、可用性监控),每个Server独立部署在隔离网络里。数据流向:外部AI → Gateway(认证)→ 路由到对应Server → 执行工具 → 返回结果 → 审计日志。全程数据不出内网,满足金融/医疗的合规要求。监控重点是工具调用成功率和P99延迟,超过阈值自动告警。"
📚 参考:MCP 官方文档(企业部署模式)
