#单实例多 MCP 聚合服务：两种实现方案深度对比

import config from "../config.json";
import { FastMCP } from "fastmcp";
import path from "node:path";
import { fileURLToPath } from "node:url";

const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);

const start = () => {
  config.forEach(async (mcpConfig) => {
    // 创建 FastMCP 服务器实例
    const server = new FastMCP({
      name: mcpConfig.name,
      version: mcpConfig.version as `${number}.${number}.${number}`,
    });

    // 动态导入服务模块并注册工具
    const { default: registerTools } = await import(
      path.join(__dirname, mcpConfig.entry.replace(".ts", ".mjs"))
    );
    registerTools(server);

    // 启动服务，监听指定端口
    server.start({
      transportType: "httpStream",
      httpStream: {
        host: "0.0.0.0",
        port: mcpConfig.port, // 每个服务独立端口
      },
    });
  });
};

start();

import { FastMCP } from "fastmcp";
import { z } from "zod";

export default function registerTools(server: FastMCP) {
  // 注册温度转换工具
  server.tool(
    "convert_temperature",
    "将温度在不同单位之间转换（摄氏度、华氏度、开尔文）",
    {
      temperature: z.number().describe("要转换的温度值"),
      from: z.enum(["celsius", "fahrenheit", "kelvin"]).describe("源温度单位"),
      to: z.enum(["celsius", "fahrenheit", "kelvin"]).describe("目标温度单位"),
    },
    async ({ temperature, from, to }) => {
      // 转换逻辑
      let celsius = 0;
      if (from === "celsius") celsius = temperature;
      else if (from === "fahrenheit") celsius = ((temperature - 32) * 5) / 9;
      else if (from === "kelvin") celsius = temperature - 273.15;

      let result = 0;
      if (to === "celsius") result = celsius;
      else if (to === "fahrenheit") result = (celsius * 9) / 5 + 32;
      else if (to === "kelvin") result = celsius + 273.15;

      return {
        content: [
          {
            type: "text",
            text: `${temperature}°${from} = ${result.toFixed(2)}°${to}`,
          },
        ],
      };
    },
  );
}

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { sseHandlers } from "express-mcp-handler";
import { z } from "zod";

/**
 * 批量创建 MCP 服务并返回 handlers 映射
 */
export async function createMcpServices(): Promise<Map<number, McpHandlers>> {
  const databaseClient = await getDatabaseClient();
  // 从数据库加载所有配置（包含关联的工具）
  const allConfigsWithTools = await databaseClient.getAllConfigsWithTools();

  const handlersMap = new Map<number, McpHandlers>();

  for (const config of allConfigsWithTools) {
    // 只创建状态为 RUNNING 的服务
    if (config.status !== StatusEnum.RUNNING) {
      continue;
    }

    // 创建 MCP 服务器实例
    const server = new McpServer({
      name: config.name,
      version: config.version,
      description: config.description,
    });

    // 批量注册工具
    config.tools.forEach((tool) => {
      // 解析 JSON Schema 并转换为 Zod schema
      const inputSchemaObj = JSON.parse(tool.input_schema);
      const outputSchemaObj = JSON.parse(tool.output_schema);
      const inputSchema = jsonSchemaToZod(inputSchemaObj);
      const outputSchema = jsonSchemaToZod(outputSchemaObj);

      // 注册工具
      server.registerTool(
        tool.name,
        {
          description: tool.description,
          inputSchema,
          outputSchema,
        },
        async ({ input }) => {
          // 执行工具回调逻辑
          // TODO: 实现实际的工具回调
          return {
            content: [{ type: "text", text: `测试: ${JSON.stringify(input)}` }],
          };
        },
      );
    });

    // 创建 SSE handlers
    const handlers = sseHandlers(() => server, {
      onError: (error: Error, sessionId?: string) => {
        consola.error(`[SSE][${config.name}] 错误:`, error);
      },
    });

    // 使用配置ID作为key存储handlers
    handlersMap.set(config.id!, handlers);
  }

  return handlersMap;
}

/**
 * 创建路由处理器（根据 MCP_ID 选择对应的 handler）
 */
export function createMcpRouteHandler(
  handlersMap: Map<number, McpHandlers>,
  handlerType: "get" | "post",
) {
  return (req: express.Request, res: express.Response, next: express.NextFunction) => {
    // 解析 MCP_ID
    const mcpIdHeader = req.headers.mcp_id || req.headers.MCP_ID;
    const mcpId = mcpIdHeader
      ? parseInt(typeof mcpIdHeader === "string" ? mcpIdHeader : mcpIdHeader[0], 10)
      : null;

    if (mcpId === null || Number.isNaN(mcpId)) {
      res.status(400).json({
        error: "缺少或无效的 MCP_ID header 参数",
        message: "请在请求 Header 中添加 MCP_ID 或 mcp_id 参数（数字类型）",
      });
      return;
    }

    // 从映射表中获取对应的 handlers
    const handlers = handlersMap.get(mcpId);

    if (!handlers) {
      res.status(404).json({
        error: `未找到 ID 为 ${mcpId} 的 MCP 配置`,
        message: `请检查 MCP_ID 是否正确，当前可用的 MCP ID: ${Array.from(handlersMap.keys()).join(", ")}`,
      });
      return;
    }

    // 调用对应的 handler（GET 或 POST）
    const targetHandler = handlerType === "get" ? handlers.getHandler : handlers.postHandler;
    targetHandler(req, res, next);
  };
}

// 注册 SSE 端点
app.get("/sse", createMcpRouteHandler(mcpHandlersMap, "get"));
app.post("/messages", createMcpRouteHandler(mcpHandlersMap, "post"));

/**
 * POST /api/config/:id/start - 启动服务
 */
export async function startConfigHandler(
  req: express.Request,
  res: express.Response,
  handlersMap: Map<number, McpHandlers>,
) {
  const id = validateIdParam(req);
  const databaseClient = await getDatabaseClient();

  // 更新数据库状态为 RUNNING
  await databaseClient.updateConfig(id, {
    status: StatusEnum.RUNNING,
    last_modified: getCurrentTimeString(),
  });
  await databaseClient.save();

  // 刷新 MCP 服务（重新加载所有配置）
  await refreshMcpServices(handlersMap);

  const updatedConfig = await databaseClient.getConfigWithTools(id);
  res.json(createSuccessResponse(dbToVO(updatedConfig)));
}

/**
 * 刷新 MCP 服务（重新加载所有配置）
 */
export async function refreshMcpServices(handlersMap: Map<number, McpHandlers>): Promise<void> {
  // 清空现有映射
  handlersMap.clear();

  // 重新创建所有服务
  const newHandlersMap = await createMcpServices();

  // 更新映射表
  newHandlersMap.forEach((handlers, id) => {
    handlersMap.set(id, handlers);
  });
}

/**
 * 将 JSON Schema 转换为 Zod schema
 */
function jsonSchemaToZod(schemaObj: Record<string, any>): z.ZodObject<any> {
  const shape: Record<string, z.ZodTypeAny> = {};

  for (const [key, value] of Object.entries(schemaObj)) {
    if (typeof value === "object" && value !== null) {
      const fieldSchema = value as { type?: string; description?: string };
      let zodType: z.ZodTypeAny;

      // 根据 JSON Schema 的 type 创建对应的 Zod 类型
      switch (fieldSchema.type) {
        case "string":
          zodType = z.string();
          break;
        case "number":
          zodType = z.number();
          break;
        case "integer":
          zodType = z.number().int();
          break;
        case "boolean":
          zodType = z.boolean();
          break;
        case "array":
          zodType = z.array(z.any());
          break;
        case "object":
          zodType = z.object({});
          break;
        default:
          zodType = z.any();
      }

      // 如果有 description，添加描述
      if (fieldSchema.description) {
        zodType = zodType.describe(fieldSchema.description);
      }

      shape[key] = zodType;
    } else {
      shape[key] = z.any();
    }
  }

  return z.object(shape);
}