Claude Code MCP实战：TypeScript开发到Docker部署

本教程面向有一定TypeScript基础的开发者，手把手教你构建一个生产级MCP服务器——连接PostgreSQL数据库与REST API，实现容器化部署，并通过Claude Code直接调用。学完后，你将掌握MCP服务器从零开发、Zod参数校验、错误处理、Docker打包到生产监控的完整链路。

适用人群：有Node.js/TypeScript基础、了解基本SQL的后端开发者，希望让Claude Code直接操作企业内部服务和数据的工程师。

学习目标：完成一个具备数据库查询、API代理、健康检查三大能力的MCP服务器，并通过Docker Compose一键部署。

前置条件

工具	版本要求	用途
Node.js	≥20.x	运行TypeScript MCP服务器
TypeScript	≥5.3	类型安全开发
Docker & Docker Compose	最新稳定版	容器化部署
PostgreSQL	≥15	示例数据库
Claude Code	最新版	MCP客户端
curl/httpie	任意	测试API

claude --version
# 预期输出: Claude Code v1.x.x 或更高

第一部分：MCP服务器架构设计

在动手写代码之前，我们需要明确MCP服务器的核心架构。一个生产级MCP服务器不是简单的"Hello World"，它需要处理传输层、工具注册、参数校验、错误恢复等多个维度。

1.1 整体架构

1.2 两种传输模式的选择

特性	Stdio（标准输入输出）	Streamable HTTP
适用场景	Claude Code本地使用	远程服务、团队共享
部署方式	本地进程	HTTP服务
认证	无需（本地信任）	需要OAuth/Token
性能	延迟极低	网络延迟
本教程重点	第一优先	第二部分补充

工程决策：Claude Code主要通过Stdio与MCP服务器通信。我们先实现Stdio模式，再扩展HTTP模式用于团队共享场景。

第二部分：项目初始化

2.1 创建项目结构

mkdir enterprise-mcp-server && cd enterprise-mcp-server
npm init -y

# MCP TypeScript SDK + Zod校验
npm install @modelcontextprotocol/sdk zod

# 数据库驱动
npm install pg

# 开发依赖
npm install -D typescript @types/node @types/pg tsx vitest

2.2 TypeScript配置

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "declaration": true,
    "sourceMap": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules", "dist"]
}

2.3 项目目录结构

mkdir -p src/{tools,utils,transports}
touch src/index.ts
touch src/tools/database.ts
touch src/tools/api-proxy.ts
touch src/tools/system.ts
touch src/utils/db.ts
touch src/utils/errors.ts

enterprise-mcp-server/
├── src/
│   ├── index.ts              # 服务器入口
│   ├── tools/
│   │   ├── database.ts       # 数据库工具集
│   │   ├── api-proxy.ts      # API代理工具集
│   │   └── system.ts         # 系统工具集
│   ├── utils/
│   │   ├── db.ts             # 数据库连接池
│   │   └── errors.ts         # 错误处理
│   └── transports/
│       └── http.ts           # HTTP传输（可选）
├── Dockerfile
├── docker-compose.yml
├── tsconfig.json
├── package.json
└── .env.example

第三部分：核心代码实现

3.1 错误处理工具

生产级服务器的第一要务是结构化错误处理。MCP协议要求错误以特定格式返回。

import { McpError } from "@modelcontextprotocol/sdk/types.js";

// 自定义错误码映射
export const ErrorCodes = {
  // MCP标准错误
  INVALID_PARAMS: -32602,
  INTERNAL_ERROR: -32603,
  // 自定义业务错误范围 (>= -32000)
  DB_CONNECTION_FAILED: -32001,
  DB_QUERY_ERROR: -32002,
  API_REQUEST_FAILED: -32003,
  API_TIMEOUT: -32004,
  UNAUTHORIZED: -32005,
} as const;

// 创建结构化MCP错误
export function createMcpError(
  code: number,
  message: string,
  details?: Record<string, unknown>
): McpError {
  const fullMessage = details
    ? `${message} | ${JSON.stringify(details)}`
    : message;
  return new McpError({
    code,
    message: fullMessage,
  });
}

// 包装异步工具处理函数
export function withErrorHandling<T extends unknown[]>(
  fn: (...args: T) => Promise<unknown>
) {
  return async (...args: T) => {
    try {
      return await fn(...args);
    } catch (error) {
      if (error instanceof McpError) throw error;

      const message =
        error instanceof Error ? error.message : String(error);
      throw createMcpError(
        ErrorCodes.INTERNAL_ERROR,
        `Unexpected error: ${message}`
      );
    }
  };
}

设计要点： - 错误码 -32001 到 -32099 是MCP规范允许的自定义范围 - withErrorHandling 高阶函数统一捕获所有工具的异常 - 错误信息包含上下文详情，方便Claude Code诊断问题

3.2 数据库连接池

import pg from "pg";

const { Pool } = pg;

let pool: pg.Pool | null = null;

export function getPool(): pg.Pool {
  if (!pool) {
    pool = new Pool({
      host: process.env.DB_HOST || "localhost",
      port: parseInt(process.env.DB_PORT || "5432"),
      database: process.env.DB_NAME || "postgres",
      user: process.env.DB_USER || "postgres",
      password: process.env.DB_PASSWORD || "",
      // 连接池配置
      max: 10,                    // 最大连接数
      idleTimeoutMillis: 30000,   // 空闲超时30秒
      connectionTimeoutMillis: 5000, // 连接超时5秒
    });

    // 监听连接错误
    pool.on("error", (err) => {
      console.error("[DB] Unexpected pool error:", err.message);
    });
  }
  return pool;
}

// 安全执行SQL查询
export async function executeQuery<T extends pg.QueryResultRow>(
  sql: string,
  params: unknown[] = []
): Promise<T[]> {
  const client = await getPool().connect();
  try {
    const result = await client.query<T>(sql, params);
    return result.rows;
  } finally {
    client.release();
  }
}

// 检查数据库连接
export async function checkConnection(): Promise<boolean> {
  try {
    await executeQuery("SELECT 1");
    return true;
  } catch {
    return false;
  }
}

// 优雅关闭
export async function closePool(): Promise<void> {
  if (pool) {
    await pool.end();
    pool = null;
  }
}

关键设计： - 连接池限制10个连接，避免耗尽数据库资源 - executeQuery 使用 connect() + release() 确保连接始终回收 - finally 块保证即使查询失败也释放连接

3.3 数据库工具集

这是整个MCP服务器的核心——让Claude Code能安全地查询数据库。

import { z } from "zod";
import { executeQuery, checkConnection } from "../utils/db.js";
import { ErrorCodes, createMcpError, withErrorHandling } from "../utils/errors.js";

// SQL注入防护：拒绝危险关键词
const DANGEROUS_PATTERNS = [
  /\bDROP\b/i,
  /\bTRUNCATE\b/i,
  /\bDELETE\s+FROM\b(?!.*\bWHERE\b)/i,  // 无WHERE的DELETE
  /\bUPDATE\b(?!.*\bWHERE\b)/i,           // 无WHERE的UPDATE
  /\bALTER\b/i,
  /\bCREATE\s+USER\b/i,
  /\bGRANT\b/i,
  /\bEXECUTE\s+IMMEDIATE\b/i,
];

function validateSqlSafety(sql: string): void {
  for (const pattern of DANGEROUS_PATTERNS) {
    if (pattern.test(sql)) {
      throw createMcpError(
        ErrorCodes.INVALID_PARAMS,
        `SQL contains potentially dangerous operation: ${pattern.source}`,
        { sql_snippet: sql.substring(0, 100) }
      );
    }
  }
}

// 工具定义
export const databaseTools = [
  {
    name: "db_query",
    description:
      "Execute a read-only SQL query on the PostgreSQL database. " +
      "Only SELECT queries are recommended. Results are limited to 500 rows.",
    inputSchema: {
      sql: z.string().describe("The SQL query to execute"),
      params: z
        .array(z.union([z.string(), z.number(), z.boolean(), z.null()]))
        .optional()
        .default([])
        .describe("Query parameters for parameterized queries"),
    },
    handler: withErrorHandling(
      async (input: { sql: string; params?: unknown[] }) => {
        validateSqlSafety(input.sql);

        // 自动添加LIMIT保护
        const sql = /\bLIMIT\b/i.test(input.sql)
          ? input.sql
          : `${input.sql.replace(/;$/, "")} LIMIT 500`;

        const rows = await executeQuery(sql, input.params);

        return {
          content: [
            {
              type: "text" as const,
              text: JSON.stringify(
                {
                  row_count: rows.length,
                  truncated: rows.length >= 500,
                  rows,
                },
                null,
                2
              ),
            },
          ],
        };
      }
    ),
  },
  {
    name: "db_describe_table",
    description:
      "Get the schema of a database table including column names, types, " +
      "constraints, and indexes.",
    inputSchema: {
      table_name: z
        .string()
        .describe("Name of the table to describe"),
      schema: z
        .string()
        .optional()
        .default("public")
        .describe("Schema name (default: public)"),
    },
    handler: withErrorHandling(
      async (input: { table_name: string; schema?: string }) => {
        const schema = input.schema || "public";

        // 查询列信息
        const columns = await executeQuery(
          `SELECT column_name, data_type, is_nullable, column_default,
                  character_maximum_length
           FROM information_schema.columns
           WHERE table_name = $1 AND table_schema = $2
           ORDER BY ordinal_position`,
          [input.table_name, schema]
        );

        // 查询索引信息
        const indexes = await executeQuery(
          `SELECT indexname, indexdef
           FROM pg_indexes
           WHERE tablename = $1 AND schemaname = $2`,
          [input.table_name, schema]
        );

        return {
          content: [
            {
              type: "text" as const,
              text: JSON.stringify(
                {
                  table: input.table_name,
                  schema,
                  columns,
                  indexes,
                },
                null,
                2
              ),
            },
          ],
        };
      }
    ),
  },
  {
    name: "db_list_tables",
    description: "List all tables in the database with row counts.",
    inputSchema: {
      schema: z
        .string()
        .optional()
        .default("public")
        .describe("Schema name (default: public)"),
    },
    handler: withErrorHandling(
      async (input: { schema?: string }) => {
        const schema = input.schema || "public";
        const tables = await executeQuery(
          `SELECT schemaname, tablename,
                  pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size
           FROM pg_tables
           WHERE schemaname = $1
           ORDER BY tablename`,
          [schema]
        );
        return {
          content: [
            {
              type: "text" as const,
              text: JSON.stringify(
                { schema, table_count: tables.length, tables },
                null,
                2
              ),
            },
          ],
        };
      }
    ),
  },
  {
    name: "db_explain",
    description:
      "Run EXPLAIN ANALYZE on a query to get the execution plan. " +
      "Useful for performance debugging.",
    inputSchema: {
      sql: z.string().describe("The SQL query to explain"),
    },
    handler: withErrorHandling(async (input: { sql: string }) => {
      validateSqlSafety(input.sql);
      const rows = await executeQuery(
        `EXPLAIN (FORMAT JSON, ANALYZE false, COSTS true) ${input.sql}`
      );
      return {
        content: [
          {
            type: "text" as const,
            text: JSON.stringify(rows[0], null, 2),
          },
        ],
      };
    }),
  },
];

这里有三层防护： 1. SQL注入防护：正则拦截危险操作模式 2. 结果集限制：自动添加 LIMIT 500，防止返回海量数据淹没上下文窗口 3. 参数化查询：所有用户输入通过 $1, $2 参数传递，杜绝经典SQL注入

3.4 API代理工具集

让Claude Code通过MCP服务器代理访问外部REST API，统一管理认证和错误处理。

import { z } from "zod";
import { ErrorCodes, createMcpError, withErrorHandling } from "../utils/errors.js";

// API请求配置
interface ApiConfig {
  baseUrl: string;
  defaultHeaders?: Record<string, string>;
  timeout?: number;
}

function getApiConfig(): ApiConfig {
  return {
    baseUrl: process.env.API_BASE_URL || "https://api.example.com",
    defaultHeaders: {
      "Content-Type": "application/json",
      ...(process.env.API_KEY
        ? { Authorization: `Bearer ${process.env.API_KEY}` }
        : {}),
    },
    timeout: parseInt(process.env.API_TIMEOUT || "10000"),
  };
}

// 通用HTTP请求函数
async function makeRequest(
  method: string,
  path: string,
  body?: unknown
): Promise<unknown> {
  const config = getApiConfig();
  const url = `${config.baseUrl}${path}`;

  const controller = new AbortController();
  const timeout = setTimeout(
    () => controller.abort(),
    config.timeout
  );

  try {
    const response = await fetch(url, {
      method,
      headers: config.defaultHeaders,
      body: body ? JSON.stringify(body) : undefined,
      signal: controller.signal,
    });

    if (!response.ok) {
      const errorBody = await response.text().catch(() => "unknown");
      throw createMcpError(
        ErrorCodes.API_REQUEST_FAILED,
        `API request failed: ${response.status} ${response.statusText}`,
        {
          url,
          method,
          status: response.status,
          response: errorBody.substring(0, 500),
        }
      );
    }

    const contentType = response.headers.get("content-type");
    if (contentType?.includes("application/json")) {
      return await response.json();
    }
    return await response.text();
  } catch (error) {
    if (error instanceof DOMException && error.name === "AbortError") {
      throw createMcpError(ErrorCodes.API_TIMEOUT, `Request timed out after ${config.timeout}ms`, {
        url,
        method,
      });
    }
    throw error;
  } finally {
    clearTimeout(timeout);
  }
}

export const apiProxyTools = [
  {
    name: "api_get",
    description:
      "Send a GET request to the configured REST API. " +
      "Use this to fetch data from external services.",
    inputSchema: {
      path: z
        .string()
        .describe("API path (e.g., /users/123)"),
      query: z
        .record(z.string())
        .optional()
        .describe("Query parameters"),
    },
    handler: withErrorHandling(
      async (input: { path: string; query?: Record<string, string> }) => {
        const queryString = input.query
          ? `?${new URLSearchParams(input.query).toString()}`
          : "";
        const result = await makeRequest("GET", `${input.path}${queryString}`);

        return {
          content: [
            {
              type: "text" as const,
              text: JSON.stringify(result, null, 2),
            },
          ],
        };
      }
    ),
  },
  {
    name: "api_post",
    description:
      "Send a POST request to the configured REST API. " +
      "Use this to create resources or trigger actions.",
    inputSchema: {
      path: z.string().describe("API path"),
      body: z
        .record(z.unknown())
        .describe("Request body as JSON object"),
    },
    handler: withErrorHandling(
      async (input: { path: string; body: Record<string, unknown> }) => {
        const result = await makeRequest("POST", input.path, input.body);
        return {
          content: [
            {
              type: "text" as const,
              text: JSON.stringify(result, null, 2),
            },
          ],
        };
      }
    ),
  },
  {
    name: "api_health",
    description:
      "Check the health status of the configured REST API endpoint.",
    inputSchema: {},
    handler: withErrorHandling(async () => {
      const config = getApiConfig();
      const start = Date.now();
      try {
        await makeRequest("GET", "/health");
        return {
          content: [
            {
              type: "text" as const,
              text: JSON.stringify({
                status: "healthy",
                url: config.baseUrl,
                latency_ms: Date.now() - start,
              }),
            },
          ],
        };
      } catch {
        return {
          content: [
            {
              type: "text" as const,
              text: JSON.stringify({
                status: "unhealthy",
                url: config.baseUrl,
                latency_ms: Date.now() - start,
              }),
            },
          ],
          isError: true,
        };
      }
    }),
  },
];

3.5 系统工具集

import { z } from "zod";
import { checkConnection } from "../utils/db.js";

export const systemTools = [
  {
    name: "system_ping",
    description: "Check if the MCP server is alive and responsive.",
    inputSchema: {},
    handler: async () => ({
      content: [
        {
          type: "text" as const,
          text: JSON.stringify({
            status: "ok",
            timestamp: new Date().toISOString(),
            version: process.env.npm_package_version || "1.0.0",
          }),
        },
      ],
    }),
  },
  {
    name: "system_status",
    description:
      "Get comprehensive status of the MCP server and all connected services.",
    inputSchema: {},
    handler: async () => {
      const dbHealthy = await checkConnection();
      return {
        content: [
          {
            type: "text" as const,
            text: JSON.stringify(
              {
                server: {
                  status: "ok",
                  uptime_seconds: process.uptime(),
                  memory_mb: Math.round(
                    process.memoryUsage().heapUsed / 1024 / 1024
                  ),
                },
                database: {
                  status: dbHealthy ? "connected" : "disconnected",
                },
                api: {
                  configured: !!process.env.API_BASE_URL,
                  base_url: process.env.API_BASE_URL || "not set",
                },
              },
              null,
              2
            ),
          },
        ],
      };
    },
  },
];

3.6 服务器主入口

#!/usr/bin/env node

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { databaseTools } from "./tools/database.js";
import { apiProxyTools } from "./tools/api-proxy.js";
import { systemTools } from "./tools/system.js";
import { closePool } from "./utils/db.js";

async function main() {
  // 创建MCP服务器实例
  const server = new McpServer({
    name: "enterprise-tools",
    version: "1.0.0",
  });

  // 注册所有工具
  const allTools = [...databaseTools, ...apiProxyTools, ...systemTools];

  for (const tool of allTools) {
    server.tool(
      tool.name,
      tool.description,
      tool.inputSchema,
      async (input: Record<string, unknown>) => {
        return await tool.handler(input);
      }
    );
  }

  // 创建Stdio传输层并连接
  const transport = new StdioServerTransport();
  await server.connect(transport);

  // 优雅关闭
  const cleanup = async () => {
    console.error("[MCP] Shutting down gracefully...");
    await closePool();
    process.exit(0);
  };

  process.on("SIGINT", cleanup);
  process.on("SIGTERM", cleanup);

  console.error("[MCP] Enterprise MCP server started on stdio");
}

main().catch((error) => {
  console.error("[MCP] Fatal error:", error);
  process.exit(1);
});

注意：console.error 而非 console.log——因为Stdio模式下标准输出用于MCP协议通信，日志必须写入stderr。

3.7 package.json脚本配置

{
  "name": "enterprise-mcp-server",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "dev": "tsx src/index.ts",
    "build": "tsc",
    "start": "node dist/index.js",
    "test": "vitest"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.12.0",
    "pg": "^8.13.0",
    "zod": "^3.24.0"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "@types/pg": "^8.11.0",
    "tsx": "^4.19.0",
    "typescript": "^5.7.0",
    "vitest": "^3.0.0"
  }
}

第四部分：连接到Claude Code

4.1 配置MCP服务器

# 数据库配置
DB_HOST=localhost
DB_PORT=5432
DB_NAME=myapp
DB_USER=postgres
DB_PASSWORD=your_password_here

# API配置（可选）
API_BASE_URL=https://api.example.com
API_KEY=your_api_key_here
API_TIMEOUT=10000

4.2 添加到Claude Code

{
  "mcpServers": {
    "enterprise-tools": {
      "command": "npx",
      "args": ["tsx", "src/index.ts"],
      "env": {
        "DB_HOST": "localhost",
        "DB_PORT": "5432",
        "DB_NAME": "myapp",
        "DB_USER": "postgres",
        "DB_PASSWORD": "your_password"
      }
    }
  }
}

claude mcp add enterprise-tools \
  --scope project \
  -e DB_HOST=localhost \
  -e DB_PORT=5432 \
  -e DB_NAME=myapp \
  -e DB_USER=postgres \
  -e DB_PASSWORD=your_password \
  -- npx tsx src/index.ts

4.3 验证连接

请用 system_ping 工具检查MCP服务器是否在线

{
  "status": "ok",
  "timestamp": "2026-05-19T08:30:00.000Z",
  "version": "1.0.0"
}

请用 db_list_tables 工具列出数据库中的所有表

4.4 常见连接问题排查

错误现象	原因	解决方案
`MCP server failed to start`	TypeScript编译错误	运行 `npx tsx src/index.ts` 检查报错
`connection refused`	数据库未启动	确认PostgreSQL运行中：`pg_isready`
`ECONNREFUSED 127.0.0.1:5432`	端口或host配置错误	检查 `.env` 中的数据库连接参数
`timeout of 5000ms exceeded`	数据库连接超时	检查网络连通性和防火墙规则
工具未出现在Claude Code中	注册失败	检查 `server.tool()` 调用的参数格式

调试技巧：在开发阶段，可以用MCP Inspector独立测试服务器：

# 安装MCP Inspector
npx @modelcontextprotocol/inspector npx tsx src/index.ts

这会启动一个Web界面（默认 http://localhost:6274），可以独立调用每个工具并查看请求/响应。

第五部分：Docker容器化部署

当MCP服务器需要在团队中共享或部署到远程环境时，Docker容器化是最佳选择。

5.1 多阶段Dockerfile

# ---- 构建阶段 ----
FROM node:22-slim AS builder

WORKDIR /app

# 先复制依赖文件，利用Docker缓存
COPY package.json package-lock.json ./
RUN npm ci

# 复制源码并构建
COPY tsconfig.json ./
COPY src/ ./src/
RUN npm run build

# ---- 运行阶段 ----
FROM node:22-slim AS runner

WORKDIR /app

# 仅安装生产依赖
COPY package.json package-lock.json ./
RUN npm ci --omit=dev && npm cache clean --force

# 从构建阶段复制编译产物
COPY --from=builder /app/dist ./dist

# 非root用户运行
RUN groupadd -r mcp && useradd -r -g mcp mcp
USER mcp

EXPOSE 3000

# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
  CMD node -e "process.exit(0)"

CMD ["node", "dist/index.js"]

关键点： - 多阶段构建减小镜像体积（不含TypeScript编译器和开发依赖） - 非 root 用户运行，满足安全合规要求 - HEALTHCHECK 指令让Docker能自动检测服务健康状态

5.2 Docker Compose编排

version: "3.9"

services:
  # PostgreSQL数据库
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: myapp
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: ${DB_PASSWORD:-changeme}
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 3s
      retries: 10

  # MCP服务器
  mcp-server:
    build: .
    environment:
      DB_HOST: postgres
      DB_PORT: 5432
      DB_NAME: myapp
      DB_USER: postgres
      DB_PASSWORD: ${DB_PASSWORD:-changeme}
      API_BASE_URL: ${API_BASE_URL:-}
      API_KEY: ${API_KEY:-}
    depends_on:
      postgres:
        condition: service_healthy
    restart: unless-stopped

volumes:
  pgdata:

-- 示例表结构
CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(20) DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT NOW()
);

-- 插入示例数据
INSERT INTO users (username, email) VALUES
    ('alice', '[email protected]'),
    ('bob', '[email protected]'),
    ('charlie', '[email protected]');

INSERT INTO orders (user_id, amount, status) VALUES
    (1, 99.99, 'completed'),
    (1, 49.50, 'pending'),
    (2, 199.00, 'completed'),
    (3, 75.25, 'shipped');

5.3 一键启动

# 构建并启动所有服务
docker compose up -d --build

# 查看日志
docker compose logs -f mcp-server

# 验证数据库初始化
docker compose exec postgres psql -U postgres -c "SELECT * FROM users;"

5.4 连接Docker化的MCP服务器

{
  "mcpServers": {
    "enterprise-tools": {
      "command": "docker",
      "args": [
        "compose", "run", "--rm", "mcp-server"
      ],
      "env": {}
    }
  }
}

Docker Compose会自动处理网络和数据库连接。Claude Code通过Docker stdin/stdout与MCP服务器通信。

第六部分：Streamable HTTP传输（团队共享模式）

当团队中多人需要使用同一个MCP服务器时，Stdio模式不再适用。我们需要HTTP传输。

6.1 HTTP传输适配器

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamablehttp.js";
import { createServer } from "http";
import { randomUUID } from "crypto";

export async function startHttpServer(
  mcpServer: McpServer,
  port: number = 3000
) {
  const httpServer = createServer(async (req, res) => {
    // CORS处理
    res.setHeader("Access-Control-Allow-Origin", "*");
    res.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS");
    res.setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization");

    if (req.method === "OPTIONS") {
      res.writeHead(204);
      res.end();
      return;
    }

    // MCP端点
    if (req.url === "/mcp" && req.method === "POST") {
      // 简单Token认证
      const authHeader = req.headers.authorization;
      const token = process.env.MCP_AUTH_TOKEN;
      if (token && authHeader !== `Bearer ${token}`) {
        res.writeHead(401, { "Content-Type": "application/json" });
        res.end(JSON.stringify({ error: "Unauthorized" }));
        return;
      }

      const transport = new StreamableHTTPServerTransport({
        sessionIdGenerator: randomUUID,
      });

      await mcpServer.connect(transport);
      await transport.handleRequest(req, res);
      return;
    }

    // 健康检查端点
    if (req.url === "/health" && req.method === "GET") {
      res.writeHead(200, { "Content-Type": "application/json" });
      res.end(JSON.stringify({ status: "ok" }));
      return;
    }

    res.writeHead(404);
    res.end("Not Found");
  });

  httpServer.listen(port, () => {
    console.error(`[MCP] HTTP server listening on port ${port}`);
  });
}

6.2 双模式启动入口

// 在 main() 函数末尾添加
const transportMode = process.env.MCP_TRANSPORT || "stdio";

if (transportMode === "http") {
  const port = parseInt(process.env.MCP_PORT || "3000");
  await startHttpServer(server, port);
} else {
  // 默认Stdio模式（原有代码）
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

6.3 HTTP模式Claude Code配置

{
  "mcpServers": {
    "enterprise-tools-remote": {
      "type": "http",
      "url": "http://your-server:3000/mcp",
      "headers": {
        "Authorization": "Bearer your-secret-token"
      }
    }
  }
}

第七部分：单元测试

import { describe, it, expect, vi, beforeEach } from "vitest";

// Mock数据库模块
vi.mock("../../utils/db.js", () => ({
  executeQuery: vi.fn(),
  checkConnection: vi.fn().mockResolvedValue(true),
}));

import { executeQuery } from "../../utils/db.js";
import { databaseTools } from "../database.js";

const mockedExecuteQuery = vi.mocked(executeQuery);

describe("database tools", () => {
  beforeEach(() => {
    vi.clearAllMocks();
  });

  describe("db_query", () => {
    const queryTool = databaseTools.find((t) => t.name === "db_query")!;

    it("should execute a simple SELECT query", async () => {
      mockedExecuteQuery.mockResolvedValueOnce([
        { id: 1, name: "test" },
      ]);

      const result = await queryTool.handler({
        sql: "SELECT * FROM users WHERE id = $1",
        params: [1],
      });

      expect(result.content[0].type).toBe("text");
      const parsed = JSON.parse(
        (result.content[0] as { text: string }).text
      );
      expect(parsed.row_count).toBe(1);
      expect(parsed.rows[0].name).toBe("test");
    });

    it("should reject dangerous DROP statements", async () => {
      await expect(
        queryTool.handler({ sql: "DROP TABLE users" })
      ).rejects.toThrow("dangerous operation");
    });

    it("should reject DELETE without WHERE clause", async () => {
      await expect(
        queryTool.handler({ sql: "DELETE FROM users" })
      ).rejects.toThrow("dangerous operation");
    });

    it("should auto-add LIMIT when missing", async () => {
      mockedExecuteQuery.mockResolvedValueOnce([]);

      await queryTool.handler({ sql: "SELECT * FROM users" });

      expect(mockedExecuteQuery).toHaveBeenCalledWith(
        expect.stringContaining("LIMIT 500"),
        []
      );
    });
  });

  describe("db_describe_table", () => {
    const describeTool = databaseTools.find(
      (t) => t.name === "db_describe_table"
    )!;

    it("should return table schema", async () => {
      mockedExecuteQuery
        .mockResolvedValueOnce([
          { column_name: "id", data_type: "integer" },
        ])
        .mockResolvedValueOnce([
          { indexname: "users_pkey", indexdef: "..." },
        ]);

      const result = await describeTool.handler({
        table_name: "users",
      });

      const parsed = JSON.parse(
        (result.content[0] as { text: string }).text
      );
      expect(parsed.table).toBe("users");
      expect(parsed.columns).toHaveLength(1);
    });
  });
});

 ✓ src/tools/__tests__/database.test.ts (4 tests) 15ms
   ✓ database tools > db_query > should execute a simple SELECT query
   ✓ database tools > db_query > should reject dangerous DROP statements
   ✓ database tools > db_query > should reject DELETE without WHERE clause
   ✓ database tools > db_query > should auto-add LIMIT when missing

 Test Files  1 passed (1)
 Tests       4 passed (4)

第八部分：生产监控与可观测性

8.1 添加结构化日志

type LogLevel = "debug" | "info" | "warn" | "error";

function formatLog(level: LogLevel, message: string, data?: unknown) {
  const entry = {
    timestamp: new Date().toISOString(),
    level,
    message,
    ...(data !== undefined ? { data } : {}),
  };
  // 所有日志输出到stderr，避免干扰MCP协议
  process.stderr.write(JSON.stringify(entry) + "\n");
}

export const logger = {
  debug: (msg: string, data?: unknown) => formatLog("debug", msg, data),
  info: (msg: string, data?: unknown) => formatLog("info", msg, data),
  warn: (msg: string, data?: unknown) => formatLog("warn", msg, data),
  error: (msg: string, data?: unknown) => formatLog("error", msg, data),
};

8.2 工具调用埋点

// 在 src/index.ts 中包装工具注册
import { logger } from "./utils/logger.js";

const TOOL_METRICS = new Map<string, { calls: number; errors: number; totalMs: number }>();

function instrumentedHandler(
  name: string,
  handler: (input: Record<string, unknown>) => Promise<unknown>
) {
  return async (input: Record<string, unknown>) => {
    const start = Date.now();
    const metrics = TOOL_METRICS.get(name) || { calls: 0, errors: 0, totalMs: 0 };

    try {
      const result = await handler(input);
      metrics.calls++;
      metrics.totalMs += Date.now() - start;
      TOOL_METRICS.set(name, metrics);

      logger.info("tool_completed", {
        tool: name,
        duration_ms: Date.now() - start,
      });

      return result;
    } catch (error) {
      metrics.errors++;
      TOOL_METRICS.set(name, metrics);

      logger.error("tool_failed", {
        tool: name,
        error: error instanceof Error ? error.message : String(error),
        duration_ms: Date.now() - start,
      });
      throw error;
    }
  };
}

// 注册时使用包装器
for (const tool of allTools) {
  server.tool(
    tool.name,
    tool.description,
    tool.inputSchema,
    instrumentedHandler(tool.name, tool.handler)
  );
}

8.3 Grafana + Prometheus集成

// 简单的Prometheus指标暴露（仅HTTP模式）
import { createServer } from "http";

function metricsHandler(req: any, res: any) {
  const lines: string[] = [];

  for (const [name, m] of TOOL_METRICS) {
    lines.push(`mcp_tool_calls_total{tool="${name}"} ${m.calls}`);
    lines.push(`mcp_tool_errors_total{tool="${name}"} ${m.errors}`);
    lines.push(`mcp_tool_duration_ms_total{tool="${name}"} ${m.totalMs}`);
  }

  res.writeHead(200, { "Content-Type": "text/plain" });
  res.end(lines.join("\n"));
}

第九部分：完整工作流程

实战案例：用Claude Code + MCP分析电商数据

假设你已完成了上述MCP服务器的开发与部署。下面展示Claude Code如何通过你的MCP服务器完成一个真实的数据分析任务。

场景

产品经理说："帮我分析一下我们最近的订单情况，看看哪些用户消费最多。"

Claude Code交互过程

用户: 帮我分析一下订单数据，找出消费最多的用户和订单状态分布

Claude Code: 我先来了解一下数据库结构，然后进行分析。

> 调用 db_list_tables
  返回: users, orders 两个表

> 调用 db_describe_table(table_name: "orders")
  返回: orders表有 id, user_id, amount, status, created_at 列

> 调用 db_query(sql: "SELECT u.username, SUM(o.amount) as total_spent,
           COUNT(o.id) as order_count FROM orders o
           JOIN users u ON u.id = o.user_id
           GROUP BY u.username ORDER BY total_spent DESC")
  返回:
  | username | total_spent | order_count |
  |----------|-------------|-------------|
  | alice    | 149.49      | 2           |
  | bob      | 199.00      | 1           |
  | charlie  | 75.25       | 1           |

> 调用 db_query(sql: "SELECT status, COUNT(*) as count,
           SUM(amount) as total FROM orders GROUP BY status")
  返回:
  | status    | count | total  |
  |-----------|-------|--------|
  | completed | 2     | 298.99 |
  | pending   | 1     | 49.50  |
  | shipped   | 1     | 75.25  |

分析结果：
1. Alice 是消费最高的用户（$149.49），下了2笔订单
2. Bob 紧随其后（$199.00），但只下了1笔大单
3. 67%的订单金额已完成，33%仍在流转中
4. 建议关注那笔pending的订单（$49.50），可能需要跟进

Claude Code自动完成了：发现表结构 → 分析字段 → 构造JOIN查询 → 聚合统计 → 生成洞察。整个过程你无需写一行SQL。

安全最佳实践清单

常见问题与解决方案

Q1: Claude Code提示"Too many tools registered"

MCP协议对工具数量有上限（通常128个）。如果你的服务器注册了过多工具，考虑： - 按功能域拆分为多个MCP服务器 - 使用工具分组，只注册高频使用的工具

Q2: 数据库查询超时导致Claude Code卡住

安全层级	措施	本教程实现
SQL注入	参数化查询 + 危险模式拦截	`validateSqlSafety()`
数据泄露	结果集大小限制	自动 `LIMIT 500`
未授权访问	Bearer Token认证	HTTP模式 `Authorization` header
容器逃逸	非root用户	Dockerfile `USER mcp`
密钥泄露	环境变量注入	`.env` + Docker Compose `env`
供应链攻击	依赖锁定	`package-lock.json` + `npm ci`
拒绝服务	查询超时	`connectionTimeoutMillis: 5000`
数据篡改	只读查询建议	危险SQL模式拦截

// 在 executeQuery 中添加语句级超时
const result = await client.query<T>(
  `SET statement_timeout = '5000'; ${sql}`,
  params
);

Q3: Docker容器中MCP服务器无法连接数据库

# 确认容器在同一Docker网络中
docker compose exec mcp-server ping postgres

# 检查环境变量是否正确传递
docker compose exec mcp-server env | grep DB_

Q4: HTTP模式下Session管理

Streamable HTTP传输会为每个连接分配Session ID。如果需要无状态模式：

// 禁用Session（每次请求创建新传输）
const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: undefined, // 无状态模式
});

Q5: 如何让Claude Code生成更准确的SQL

description: `Execute SQL on a PostgreSQL database.
Tables: users(id, username, email, created_at), orders(id, user_id, amount, status, created_at).
Always use parameterized queries ($1, $2) for user inputs.`

工具描述会作为System Prompt的一部分发送给Claude，更详细的描述意味着更准确的SQL生成。

Claude Code MCP服务器TypeScript实战：从开发到Docker部署

概述