MCP协议高阶实战：构建企业级带认证的流式Server

import jwt from 'jsonwebtoken';

const SECRET_KEY = process.env.JWT_SECRET || 'your-super-secret-key';

export const authenticateToken = (req, res, next) => {
  const authHeader = req.headers['authorization'];
  const token = authHeader && authHeader.split(' ')[1];

  if (!token) return res.status(401).json({ error: 'Missing token' });

  jwt.verify(token, SECRET_KEY, (err, user) => {
    if (err) return res.status(403).json({ error: 'Invalid token' });
    req.user = user; // 将用户信息注入请求上下文
    next();
  });
};

import express from 'express';
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';

const app = express();
const mcpServer = new Server({ name: 'Enterprise-MCP', version: '1.0.0' }, {
  capabilities: { tools: {}, resources: {} }
});

// 存储活跃的传输通道
const transports = new Map();

// 1. 建立 SSE 连接 (需要鉴权)
app.get('/sse', authenticateToken, async (req, res) => {
  const transport = new SSEServerTransport('/message', res);
  
  // 将 userId 绑定到特定的 sessionId
  const sessionId = req.query.sessionId || crypto.randomUUID();
  transport.userId = req.user.id; 
  transports.set(sessionId, transport);

  await mcpServer.connect(transport);
  
  req.on('close', () => {
    transports.delete(sessionId);
  });
});

// 2. 接收客户端消息 (同样需要鉴权)
app.post('/message', authenticateToken, express.json(), async (req, res) => {
  const sessionId = req.query.sessionId;
  const transport = transports.get(sessionId);

  if (!transport) {
    return res.status(404).send('Session not found');
  }
  
  // 安全校验：确保发送消息的用户与建立连接的用户一致
  if (transport.userId !== req.user.id) {
    return res.status(403).send('Access denied for this session');
  }

  await transport.handlePostMessage(req, res);
});

// 注册一个分页读取大日志文件的 Tool
mcpServer.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === 'read_large_log') {
    const { offset = 0, limit = 1000 } = request.params.arguments;
    
    // 使用流读取特定范围的数据，避免将整个文件读入内存
    const logChunk = await streamFileChunk('/var/log/enterprise.log', offset, limit);
    
    // 验证返回的 JSON 结构是否符合规范
    // (在调试阶段，可以使用 QubitTool 的 JSON 格式化工具进行检查)
    return {
      content: [
        {
          type: 'text',
          text: JSON.stringify({
            data: logChunk,
            nextOffset: offset + limit,
            hasMore: logChunk.length === limit
          })
        }
      ]
    };
  }
  throw new Error('Tool not found');
});