<!-- title: Claude Code MCP实战：TypeScript开发到Docker部署 -->

# 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 Code并能正常运行：

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

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

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

### 1.1 整体架构


![图表1](images/claude-code-tutorial-r3-1257-mermaid-0.png)

> *图表 1*


### 1.2 两种传输模式的选择

MCP服务器支持两种传输方式：

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

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

## 第二部分：项目初始化

### 2.1 创建项目结构

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

安装核心依赖：

```bash
# 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配置

创建 `tsconfig.json`：

```json
{
  "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 项目目录结构

```bash
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协议要求错误以特定格式返回。

`src/utils/errors.ts`：

```typescript
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 数据库连接池

`src/utils/db.ts`：

```typescript
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能安全地查询数据库。

`src/tools/database.ts`：

```typescript
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),
          },
        ],
      };
    }),
  },
];
```

**安全设计深度解析**：


![图表2](images/claude-code-tutorial-r3-1257-mermaid-1.png)

> *图表 2*


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

### 3.4 API代理工具集

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

`src/tools/api-proxy.ts`：

```typescript
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 系统工具集

`src/tools/system.ts`：

```typescript
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 服务器主入口

`src/index.ts`：

```typescript
#!/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脚本配置

```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服务器

在项目根目录创建 `.env` 文件：

```bash
# 数据库配置
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

在项目根目录创建 `.mcp.json`（项目级配置）：

```json
{
  "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` 命令行方式：

```bash
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 验证连接

启动Claude Code，检查MCP服务器是否成功连接：

```bash
claude
```

在Claude Code交互界面中输入：

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

**预期输出**：
```json
{
  "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独立测试服务器：

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

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

## 第五部分：Docker容器化部署

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

### 5.1 多阶段Dockerfile

`Dockerfile`：

```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编排

`docker-compose.yml`：

```yaml
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:
```

创建示例初始化SQL `init.sql`：

```sql
-- 示例表结构
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', 'alice@example.com'),
    ('bob', 'bob@example.com'),
    ('charlie', 'charlie@example.com');

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 一键启动

```bash
# 构建并启动所有服务
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服务器

对于本地Stdio模式，更新 `.mcp.json`：

```json
{
  "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传输适配器

`src/transports/http.ts`：

```typescript
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 双模式启动入口

更新 `src/index.ts`，支持通过环境变量切换传输模式：

```typescript
// 在 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配置

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

## 第七部分：单元测试

生产级代码必须有测试。使用Vitest为MCP工具编写测试。

`src/tools/__tests__/database.test.ts`：

```typescript
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);
    });
  });
});
```

运行测试：

```bash
npx vitest run
```

**预期输出**：
```
 ✓ 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 添加结构化日志

在 `src/utils/logger.ts` 中：

```typescript
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 工具调用埋点

在工具注册时添加自动指标采集：

```typescript
// 在 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监控基础设施，可以暴露指标端点：

```typescript
// 简单的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"));
}
```

## 第九部分：完整工作流程


![图表3](images/claude-code-tutorial-r3-1257-mermaid-2.png)

> *图表 3*

## 实战案例：用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。

## 安全最佳实践清单


![图表4](images/claude-code-tutorial-r3-1257-mermaid-3.png)

> *图表 4*


| 安全层级 | 措施 | 本教程实现 |
|---------|------|-----------|
| 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模式拦截 |

## 常见问题与解决方案

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

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

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

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

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

检查网络配置：
```bash
# 确认容器在同一Docker网络中
docker compose exec mcp-server ping postgres

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

### Q4: HTTP模式下Session管理

Streamable HTTP传输会为每个连接分配Session ID。如果需要无状态模式：
```typescript
// 禁用Session（每次请求创建新传输）
const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: undefined, // 无状态模式
});
```

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

在工具描述中添加数据库上下文：
```typescript
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生成。

## 延伸阅读

- [Anthropic MCP官方文档](https://modelcontextprotocol.io/) — MCP协议规范与SDK参考
- [MCP TypeScript SDK GitHub](https://github.com/modelcontextprotocol/typescript-sdk) — 源码与示例
- [Claude Code MCP配置指南](https://code.claude.com/docs/en/mcp) — 官方MCP集成文档
- [OWASP MCP安全速查表](https://cheatsheetseries.owasp.org/cheatsheets/MCP_Security_Cheat_Sheet.html) — MCP安全最佳实践
- [Docker MCP Catalog](https://www.docker.com/blog/docker-mcp-catalog-secure-way-to-discover-and-run-mcp-servers/) — Docker官方MCP服务器目录

## 参考来源

1. [MCP TypeScript SDK官方仓库](https://github.com/modelcontextprotocol/typescript-sdk) — TypeScript SDK完整API与传输层实现
2. [Connect Claude Code to tools via MCP](https://code.claude.com/docs/en/mcp) — Claude Code MCP服务器配置官方文档
3. [Adding custom tools to an MCP server in TypeScript](https://mcpcat.io/guides/adding-custom-tools-mcp-server-typescript/) — Zod校验与错误处理实践
4. [Build a Custom MCP Server for Claude Code: Complete Guide](https://apidog.com/blog/how-to-quickly-build-a-mcp-server-for-claude-code/) — 从构建到测试的完整指南
5. [MCP Best Practices: Architecture & Implementation Guide](https://modelcontextprotocol.info/docs/best-practices/) — 生产级MCP架构最佳实践
6. [OWASP MCP Security Cheat Sheet](https://cheatsheetseries.owasp.org/cheatsheets/MCP_Security_Cheat_Sheet.html) — MCP安全攻防速查表

> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)