企业级 MCP 网关架构实践分享

当你开始构建超越简单实验的 AI 应用时，一切都会改变。模型需要访问文件、数据库、API 和内部服务。这就是模型上下文协议（MCP）的用武之地。

但在生产环境中管理数十个 MCP 服务器、工具和集成很快就会变成一场噩梦。过去几个月我使用 Bifrost 构建了一个企业级 MCP 网关，我想分享我学到的东西。

💻 问题：没有网关的 MCP 很糟糕

没有适当基础设施的情况下会发生什么：

你的模型花费宝贵的 token 来发现可用的工具。团队无法控制谁使用什么。一个工程师不小心删除了错误的数据库，因为模型有它本不应该有的访问权限。API 成本意外激增。你不知道哪些 AI 工作流运行在哪里。

根本问题是：MCP 是为灵活性设计的。当你从聊天机器人扩展到生产 AI 系统时，你需要：

集中式工具管理而不是分散的 MCP 服务器

细粒度访问控制，使营销工具不会泄露到工程部门

每个工具的速率限制以防止 API 滥用和成本失控

完整的审计日志以供合规和调试

Bifrost 是一个高性能的、基于 Go 的 LLM 网关，它解决了这些问题：

# Quick start - 30 seconds with -p 8000
npx -y @maximhq/bifrost

# Opens http://localhost:8000

比其他网关低 40 倍的开销（11µs vs 440µs）

内存使用减少 68%

在 5,000 RPS 下 100% 成功率

代码模式 - 模型生成编排代码而不是逐步调用

语义缓存 - 类似查询的成本降低 40-60%

内置控制 - RBAC、速率限制、成本追踪、审计日志

📦 1. 收集所有 MCP 服务器

与其直接给模型访问分散的 MCP 服务器：

// Gateway configuration - single entry point
mcpConfig := &schemas.MCPConfig{
    ClientConfigs: []schemas.MCPClientConfig{
        {
            Name:           "filesystem",
            ConnectionType: schemas.MCPConnectionTypeSTDIO,
            StdioConfig: &schemas.MCPStdioConfig{
                Command: "npx",
                Args:    []string{"-y", "@anthropic/mcp-filesystem"},
            },
            ToolsToExecute: []string{"*"},
        },
        {
            Name:             "web_search",
            ConnectionType:   schemas.MCPConnectionTypeHTTP,
            ConnectionString: bifrost.Ptr("http://localhost:3001/mcp"),
            ToolsToExecute:   []string{"search", "fetch_url"},
        },
    },
}

client, err := bifrost.Init(context.Background(), schemas.BifrostConfig{
    Account:   account,
    MCPConfig: mcpConfig,
    Logger:    bifrost.NewDefaultLogger(schemas.LogLevelInfo),
})

所有工具的单一信息来源

统一的安全策略

集中监控和成本追踪

所有模型的一致行为

⚙️ 2. 基于角色控制工具访问

不同的团队需要不同的工具访问级别。实施基于角色的访问控制：

roleToToolsMapping := map[string][]string{
    "engineering": {"filesystem", "database", "github-api"},
    "marketing":   {"web-search", "document-generation"},
    "finance":     {"cost-tracking"},
    "admin":       {"*"},  // All tools
}

roleLimits := map[string]map[string]int{
    "engineering": {"filesystem": 1000, "database": 500},
    "marketing":   {"web_search": 100},
    "finance":     {"cost_tracking": 50},
}

// Check access
async function checkToolAccess(userId, role, toolName) {
  const allowedTools = roleToToolsMapping[role];
  if (!allowedTools.includes(toolName)) {
    throw new Error(`Tool '${toolName}' is denied for role '${role}'`);
  }
}

实际例子 - 访问被拒绝：

curl -X POST http://localhost:8000/v1/mcp/tool/execute \
  -H "Content-Type: application/json" \
  -d '{
    "tool_call": {
      "tool_name": "database",
      "params": {"query": "SELECT * FROM users"}
    },
    "user_role": "marketing"
  }'

# Response (403):
# {
#   "error": "Access Denied",
#   "message": "Tool 'database' is not allowed for role 'marketing'"
# }

这一项改变就能防止整类安全问题。

🔎 3. 实施速率限制

一个 AI 工作流曾经陷入循环，每秒向数据库发送数千个查询。成本在 2 小时内激增了 2000 美元，直到我们发现才停止。

速率限制是你自身系统的防火墙：

class RateLimiter {
  async checkLimit(toolName, userId, limit) {
    const key = `${toolName}:${userId}`;
    const now = Date.now();
    const windowStart = now - 60000; // 1 minute

    if (!this.windows.has(key)) {
      this.windows.set(key, []);
    }

    const timestamps = this.windows.get(key)
      .filter(t => t > windowStart);

    if (timestamps.length >= limit) {
      return {
        allowed: false,
        retryAfter: Math.ceil((timestamps[0] + 60000 - now) / 1000)
      };
    }

    timestamps.push(now);
    return { allowed: true, remaining: limit - timestamps.length };
  }
}

实际例子 - 超过速率限制：

curl -X POST http://localhost:8000/v1/mcp/tool/execute \
  -H "Content-Type: application/json" \
  -d '{
    "tool_call": {
      "tool_name": "web_search",
      "params": {"query": "another search"}
    },
    "user_id": "user-123",
    "user_role": "marketing"
  }'

# Response (429 - Rate Limited):
# {
#   "error": "Rate Limit Exceeded",
#   "message": "Tool 'web_search' limit exceeded (100/min)",
#   "retryAfter": 45
# }

速率限制器在 30 秒内抓住了一个可能导致 5000 美元+ 损失的事件。

📊 4. 追踪成本和审计一切

生产 AI 系统需要问责制。谁运行了什么？什么时候？成本是多少？

type AuditLog struct {
    Timestamp  time.Time
    UserId     string
    UserRole   string
    ToolName   string
    Success    bool
    Cost       float64
    Duration   time.Duration
    Error      string
}

async function executeTool(toolName, params, context) {
  const startTime = Date.now();

  try {
    const result = await toolExecutor.execute(toolName, params);
    const duration = Date.now() - startTime;
    const cost = calculateCost(toolName, params);

    await auditLogger.log({
      userId: context.userId,
      userRole: context.userRole,
      toolName,
      success: true,
      cost,
      duration
    });

    return result;
  } catch (error) {
    await auditLogger.log({
      userId: context.userId,
      toolName,
      success: false,
      error: error.message
    });
    throw error;
  }
}

例子 - 成本分解：

GET /v1/analytics/costs?team_id=team-engineering&period=month

{
  "total_cost": "$127.45",
  "budget": "$1000.00",
  "remaining": "$872.55",
  "usage_by_tool": [
    {
      "tool": "web_search",
      "calls": 1234,
      "cost": "$12.34"
    },
    {
      "tool": "database",
      "calls": 567,
      "cost": "$56.70"
    }
  ]
}

这种可见性是变革性的。团队可以看到他们的确切花费。异常情况变得显而易见。

🖋️ 完整流程

这是有所有控制层的工具执行的样子：

app.post("/v1/mcp/tool/execute", async (req, res) => {
  const { toolName, params, userId, userRole, teamId } = req.body;

  try {
    // 1. Check role-based access
    await checkToolAccess(userId, userRole, toolName);

    // 2. Check rate limits
    const limit = roleLimits[userRole]?.[toolName];
    const rateLimitCheck = await limiter.checkLimit(toolName, userId, limit);
    if (!rateLimitCheck.allowed) {
      return res.status(429).json({
        error: "Rate Limit Exceeded",
        retryAfter: rateLimitCheck.retryAfter
      });
    }

    // 3. Check budget
    const cost = estimateCost(toolName, params);
    const budgetCheck = await budgetTracker.deductCost(teamId, toolName, cost);
    if (!budgetCheck.allowed) {
      return res.status(402).json({
        error: "Budget Exceeded"
      });
    }

    // 4. Execute tool
    const result = await executeTool(toolName, params);

    // 5. Log the action
    await auditLogger.log({
      userId, userRole, teamId, toolName,
      success: true, cost, duration
    });

    res.json({ success: true, data: result });

  } catch (error) {
    // Log failures too
    await auditLogger.log({
      userId, userRole, teamId, toolName,
      success: false, error: error.message
    });

    res.status(400).json({ success: false, error: error.message });
  }
});

与其逐个调用工具，模型生成编排它们的 TypeScript 代码：

// Model generates this automatically
const tools = await listToolFiles();  // List available tools
const githubTool = await readToolFile('github');  // Read definition

// Execute a complete workflow
const results = await executeToolCode(async () => {
  const repos = await github.search_repos({ 
    query: "golang bifrost", 
    maxResults: 5 
  });

  const formatted = repos.items.map(repo => ({
    name: repo.name,
    stars: repo.stargazers_count,
    url: repo.html_url
  }));

  return { repositories: formatted, count: formatted.length };
});

~40% 的 token 使用率降低

单次执行 vs 多次调用

更好的控制和调试

Bifrost 在 5,000 RPS 下的性能：

Goroutines：轻量级并发（~2 KB 每个）

编译二进制：无启动开销

内存高效：比其他网关少 68%

跨 CPU 核心的真正并行

# 1. Install Bifrost (30 seconds)
npx -y @maximhq/bifrost

# 2. Configure API keys (.env)
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...

# 3. Open dashboard
open http://localhost:8000

# 4. Make your first call
curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "Hello Bifrost!"}]
  }'

# 5. Drop-in replacement
# Change this:
base_url = "https://api.openai.com"
# To this:
base_url = "http://localhost:8000/openai"

💻 我会做得不同的地方

从第一天开始进行成本追踪 - 事后改造很痛苦

让速率限制可配置 - 不同团队有不同需求

积极实施缓存 - 语义缓存节省 40% 以上

构建分层权限 - 扁平模型不可扩展

设置实时警报 - 不要等待每周审查

归根结底，网关不是为了花哨。它是为了控制。

当你集中工具管理时，你会获得：

安全性 - 工具按角色隔离，错误被限制

可见性 - 每个操作都被记录并追踪成本

优化 - 看到什么昂贵并修复它

调试 - 完整的审计日志用于事件处理

对我们来说，这个基础设施把 AI 从"一个很酷的演示"变成了我们可以有信心部署到生产的东西。

Bifrost GitHub: https://github.com/maximhq/bifrost

Documentation: https://docs.getbifrost.ai

MCP Spec: https://modelcontextprotocol.io

你在大规模构建 AI 基础设施吗？在评论中告诉我！