MCP Server工程化：调试安全到生产部署

本教程面向已了解MCP基础概念的开发者，聚焦从本地开发调试、安全加固到生产部署的完整工程化流程。你将掌握MCP Inspector调试技巧、OAuth 2.1认证实现、输入验证与速率限制策略、Streamable HTTP传输配置、以及MCP Registry发布流程。学完后能独立交付一个可上线的MCP Server。

适用人群：有TypeScript基础、已了解MCP三原语（Tool/Resource/Prompt）概念的开发者。

前置条件

一、MCP协议最新状态：你需要知道的变更

在动手之前，必须了解MCP协议在2025-2026年的重要演进，这些变更直接影响你的开发决策。

1.1 三大特性已弃用

2026年4月，MCP核心维护者Kurtis Van Gent提交了SEP-2577提案，正式弃用三个低采用率特性：

弃用特性	原因	替代方案
Sampling	LLM调用应由Client控制，Server不应发起	Client侧编排Agent逻辑
Roots	用例有限，增加协议复杂度	Server自行管理文件系统
Logging	标准化日志格式收益低	Server自行实现日志（winston/pino）

影响：新项目不要依赖这三个特性。现有的server.setRequestHandler(SamplingRequestSchema, ...)等代码需要移除。

1.2 传输协议演进

时间线：
2024-11  MCP发布，仅支持stdio
2025-03  新增SSE（Server-Sent Events）远程传输
2025-06  弃用SSE，引入Streamable HTTP
2025-11  Streamable HTTP成为推荐远程传输方式
2026-03  MCP Registry上线，标准化发现与分发

核心决策：本地MCP Server用stdio（简单高效），远程Server用Streamable HTTP（支持无状态部署、负载均衡）。

1.3 MCP协议架构全景

graph TB
    subgraph "Host Application"
        CC[Claude Code CLI]
        CD[Claude Desktop]
        VS[VS Code Copilot]
    end

    subgraph "MCP Client"
        MC1[Client 1]
        MC2[Client 2]
        MC3[Client 3]
    end

    subgraph "传输层"
        STDIO[stdio<br/>本地进程通信]
        HTTP[Streamable HTTP<br/>远程HTTP]
    end

    subgraph "MCP Server"
        S1[你的自定义Server]
        S2[第三方Server]
        S3[官方示例Server]
    end

    subgraph "三原语"
        T[Tools<br/>可执行操作]
        R[Resources<br/>结构化数据]
        P[Prompts<br/>提示模板]
    end

    CC --> MC1 --> STDIO --> S1
    CD --> MC2 --> HTTP --> S2
    VS --> MC3 --> STDIO --> S3
    S1 --> T
    S1 --> R
    S1 --> P

二、搭建工程化开发环境

2.1 项目初始化

mkdir mcp-server-taskmanager && cd mcp-server-taskmanager
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node vitest

我们选择构建一个"任务管理器"MCP Server作为实战案例——它比Hello World复杂，又不过度庞大，非常适合展示工程化实践。

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", "tests"]
}

2.3 项目结构

mcp-server-taskmanager/
├── src/
│   ├── index.ts          # 入口，Server初始化
│   ├── server.ts         # MCP Server核心逻辑
│   ├── tools/            # Tool定义
│   │   ├── index.ts
│   │   ├── task-create.ts
│   │   ├── task-list.ts
│   │   └── task-complete.ts
│   ├── resources/        # Resource定义
│   │   ├── index.ts
│   │   └── task-resource.ts
│   ├── prompts/          # Prompt定义
│   │   ├── index.ts
│   │   └── daily-summary.ts
│   ├── store/            # 状态管理
│   │   └── task-store.ts
│   ├── transport/        # 传输层
│   │   ├── stdio.ts
│   │   └── http.ts
│   ├── middleware/       # 中间件
│   │   ├── rate-limit.ts
│   │   └── validator.ts
│   └── errors.ts        # 错误定义
├── tests/
│   ├── tools.test.ts
│   └── store.test.ts
├── inspector/           # Inspector配置
│   └── config.json
├── package.json
├── tsconfig.json
└── Dockerfile

2.4 package.json关键配置

{
  "name": "mcp-server-taskmanager",
  "version": "1.0.0",
  "type": "module",
  "main": "dist/index.js",
  "bin": {
    "mcp-server-taskmanager": "dist/index.js"
  },
  "scripts": {
    "build": "tsc",
    "dev": "tsc --watch",
    "start": "node dist/index.js",
    "start:http": "node dist/transport/http.js",
    "test": "vitest",
    "inspector": "npx @modelcontextprotocol/inspector node dist/index.js"
  },
  "files": ["dist"]
}

关键点： - "type": "module" — MCP SDK使用ES模块 - "bin" — 配置为可执行命令，发布后用户可直接npx运行 - "files" — 只发布编译产物，不发布源码

三、Server核心实现：带错误处理的工程化写法

3.1 自定义错误体系

export class McpTaskError extends Error {
  constructor(
    message: string,
    public readonly code: string,
    public readonly statusCode: number = 400
  ) {
    super(message);
    this.name = 'McpTaskError';
  }
}

export const ErrorCodes = {
  TASK_NOT_FOUND: 'TASK_NOT_FOUND',
  INVALID_INPUT: 'INVALID_INPUT',
  DUPLICATE_TASK: 'DUPLICATE_TASK',
  RATE_LIMITED: 'RATE_LIMITED',
  INTERNAL_ERROR: 'INTERNAL_ERROR',
} as const;

工程化原则：每个错误都有明确的code和statusCode，方便Client端做条件处理。不要用throw new Error("something went wrong")这种模糊写法。

3.2 状态管理：TaskStore

export interface Task {
  id: string;
  title: string;
  description: string;
  status: 'pending' | 'in_progress' | 'completed';
  priority: 'low' | 'medium' | 'high';
  createdAt: string;
  completedAt?: string;
}

export class TaskStore {
  private tasks = new Map<string, Task>();

  create(input: { title: string; description: string; priority: string }): Task {
    const id = `task_${Date.now()}_${Math.random().toString(36).slice(2, 7)}`;
    const task: Task = {
      id,
      title: input.title,
      description: input.description,
      status: 'pending',
      priority: this.validatePriority(input.priority),
      createdAt: new Date().toISOString(),
    };
    this.tasks.set(id, task);
    return task;
  }

  list(filter?: { status?: string; priority?: string }): Task[] {
    let result = Array.from(this.tasks.values());
    if (filter?.status) {
      result = result.filter(t => t.status === filter.status);
    }
    if (filter?.priority) {
      result = result.filter(t => t.priority === filter.priority);
    }
    return result.sort((a, b) =>
      new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()
    );
  }

  complete(id: string): Task {
    const task = this.tasks.get(id);
    if (!task) {
      throw new McpTaskError(
        `Task not found: ${id}`,
        ErrorCodes.TASK_NOT_FOUND,
        404
      );
    }
    task.status = 'completed';
    task.completedAt = new Date().toISOString();
    return task;
  }

  get(id: string): Task | undefined {
    return this.tasks.get(id);
  }

  private validatePriority(p: string): 'low' | 'medium' | 'high' {
    if (!['low', 'medium', 'high'].includes(p)) return 'medium';
    return p as 'low' | 'medium' | 'high';
  }
}

注意：这里使用内存Map。生产环境可以替换为Redis、SQLite或任何持久化方案——Store的接口不变，只换实现。

3.3 Server初始化：支持双传输

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import { registerTools } from './tools/index.js';
import { registerResources } from './resources/index.js';
import { registerPrompts } from './prompts/index.js';
import { TaskStore } from './store/task-store.js';

export function createServer(): { server: McpServer; store: TaskStore } {
  const store = new TaskStore();

  const server = new McpServer({
    name: 'task-manager',
    version: '1.0.0',
  });

  // 注册三原语
  registerTools(server, store);
  registerResources(server, store);
  registerPrompts(server, store);

  return { server, store };
}

3.4 Tool注册：带输入验证

import { z } from 'zod';
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import type { TaskStore } from '../store/task-store.js';
import { McpTaskError, ErrorCodes } from '../errors.js';

export function registerTaskCreate(server: McpServer, store: TaskStore) {
  server.tool(
    'task_create',
    'Create a new task with title, description, and priority',
    {
      title: z.string().min(1).max(200).describe('Task title'),
      description: z.string().max(2000).describe('Detailed task description'),
      priority: z.enum(['low', 'medium', 'high'])
        .default('medium')
        .describe('Task priority level'),
    },
    async (input) => {
      // Zod已做基础验证，这里做业务层验证
      if (input.title.trim().length === 0) {
        throw new McpTaskError(
          'Title cannot be empty or whitespace only',
          ErrorCodes.INVALID_INPUT
        );
      }

      const task = store.create({
        title: input.title.trim(),
        description: input.description,
        priority: input.priority,
      });

      return {
        content: [
          {
            type: 'text' as const,
            text: JSON.stringify({
              success: true,
              task,
              message: `Task "${task.title}" created with ID: ${task.id}`,
            }, null, 2),
          },
        ],
      };
    }
  );
}

工程化要点： - Zod schema同时充当参数验证和文档生成——Claude Code会读取schema描述来理解Tool用途 - 业务层验证和schema验证分离，各司其职 - 返回结构化的JSON，方便LLM解析

import { z } from 'zod';
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import type { TaskStore } from '../store/task-store.js';

export function registerTaskList(server: McpServer, store: TaskStore) {
  server.tool(
    'task_list',
    'List all tasks with optional status and priority filters',
    {
      status: z.enum(['pending', 'in_progress', 'completed'])
        .optional()
        .describe('Filter by task status'),
      priority: z.enum(['low', 'medium', 'high'])
        .optional()
        .describe('Filter by priority level'),
    },
    async (input) => {
      const tasks = store.list({
        status: input.status,
        priority: input.priority,
      });

      return {
        content: [
          {
            type: 'text' as const,
            text: JSON.stringify({
              total: tasks.length,
              tasks,
            }, null, 2),
          },
        ],
      };
    }
  );
}

import { z } from 'zod';
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import type { TaskStore } from '../store/task-store.js';

export function registerTaskComplete(server: McpServer, store: TaskStore) {
  server.tool(
    'task_complete',
    'Mark a task as completed by its ID',
    {
      id: z.string().startsWith('task_').describe('The task ID to complete'),
    },
    async (input) => {
      const task = store.complete(input.id);
      return {
        content: [
          {
            type: 'text' as const,
            text: JSON.stringify({
              success: true,
              task,
              message: `Task "${task.title}" marked as completed`,
            }, null, 2),
          },
        ],
      };
    }
  );
}

import type { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import type { TaskStore } from '../store/task-store.js';
import { registerTaskCreate } from './task-create.js';
import { registerTaskList } from './task-list.js';
import { registerTaskComplete } from './task-complete.js';

export function registerTools(server: McpServer, store: TaskStore) {
  registerTaskCreate(server, store);
  registerTaskList(server, store);
  registerTaskComplete(server, store);
}

3.5 Resource注册：动态数据源

Resource是Server暴露给LLM的结构化数据通道——不同于Tool的"执行操作"，Resource是"读取数据"。

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import type { TaskStore } from '../store/task-store.js';

export function registerTaskResources(server: McpServer, store: TaskStore) {
  // 静态Resource模板：列出所有任务
  server.resource(
    'all-tasks',
    'task://all',
    async (uri) => {
      const tasks = store.list();
      return {
        contents: [
          {
            uri: uri.href,
            mimeType: 'application/json',
            text: JSON.stringify(tasks, null, 2),
          },
        ],
      };
    }
  );

  // 动态Resource模板：按ID获取单个任务
  server.resource(
    'task-by-id',
    new URL('task:///{id}'),  // URI模板参数
    async (uri, { id }) => {
      const task = store.get(id as string);
      if (!task) {
        return {
          contents: [
            {
              uri: uri.href,
              mimeType: 'application/json',
              text: JSON.stringify({ error: 'Task not found' }),
            },
          ],
        };
      }
      return {
        contents: [
          {
            uri: uri.href,
            mimeType: 'application/json',
            text: JSON.stringify(task, null, 2),
          },
        ],
      };
    }
  );
}

3.6 Prompt注册：可复用的提示模板

import { z } from 'zod';
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import type { TaskStore } from '../store/task-store.js';

export function registerDailySummary(server: McpServer, store: TaskStore) {
  server.prompt(
    'daily-summary',
    'Generate a daily task summary report',
    {
      date: z.string().optional().describe('Date in YYYY-MM-DD format, defaults to today'),
      includeCompleted: z.boolean().default(true).describe('Whether to include completed tasks'),
    },
    async (input) => {
      const tasks = store.list();
      const pending = tasks.filter(t => t.status !== 'completed');
      const completed = tasks.filter(t => t.status === 'completed');

      let report = `# Daily Task Summary\n\n`;
      report += `**Total tasks**: ${tasks.length}\n`;
      report += `**Pending**: ${pending.length} | **Completed**: ${completed.length}\n\n`;

      if (pending.length > 0) {
        report += `## Pending Tasks\n`;
        for (const t of pending) {
          report += `- [${t.priority.toUpperCase()}] ${t.title}\n`;
        }
        report += '\n';
      }

      if (input.includeCompleted && completed.length > 0) {
        report += `## Completed Tasks\n`;
        for (const t of completed) {
          report += `- ~~${t.title}~~ (completed at ${t.completedAt})\n`;
        }
      }

      return {
        messages: [
          {
            role: 'user' as const,
            content: {
              type: 'text' as const,
              text: report,
            },
          },
        ],
      };
    }
  );
}

3.7 入口文件：双传输支持

import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { createServer } from '../server.js';

async function main() {
  const { server } = createServer();
  const transport = new StdioServerTransport();
  await server.connect(transport);
  // stdio模式下不要console.log，会污染JSON-RPC通道
}

main().catch((error) => {
  // 错误输出到stderr，不干扰stdout的MCP通信
  process.stderr.write(`Fatal error: ${error.message}\n`);
  process.exit(1);
});

// 默认使用stdio传输
import './transport/stdio.js';

四、调试：MCP Inspector实战

调试MCP Server最强大的工具是官方的MCP Inspector——一个可视化的交互式调试器。

4.1 启动Inspector

# 先编译
npm run build

# 启动Inspector + 你的Server
npx @modelcontextprotocol/inspector node dist/transport/stdio.js

Inspector会自动： 1. 启动一个本地Web UI（默认http://localhost:6274） 2. 以stdio方式启动你的MCP Server 3. 在Browser中打开调试界面

4.2 Inspector调试流程

flowchart LR
    A[启动Inspector] --> B[连接Server]
    B --> C{初始化握手}
    C -->|成功| D[查看Capabilities]
    C -->|失败| E[检查错误日志]
    E --> F[修复代码]
    F --> A
    D --> G[测试Tools]
    D --> H[测试Resources]
    D --> I[测试Prompts]
    G --> J[查看请求/响应]
    H --> J
    I --> J
    J --> K{结果正确?}
    K -->|否| F
    K -->|是| L[配置到Claude Code]

4.3 Inspector中的关键操作

Tools标签页： - 选择一个Tool，填写参数，点击"Run" - 查看完整的JSON-RPC请求和响应 - 验证Zod schema是否正确生成参数表单

Resources标签页： - 点击Resource URI查看返回内容 - 验证mimeType是否正确 - 测试动态Resource模板的参数替换

Prompts标签页： - 填写Prompt参数并渲染 - 验证返回的messages格式

Connection标签页： - 查看Server的capabilities声明 - 监控JSON-RPC消息流 - 查看错误和通知

4.4 常见调试问题

4.5 日志策略：不要用Sampling

问题	原因	解决方案
Inspector无法连接	Server启动失败	检查编译输出，手动`node dist/index.js`测试
Tool不显示	注册顺序问题	确保在`server.connect()`之前注册所有handler
参数验证失败	Zod schema错误	在Inspector中查看生成的schema，对照文档
响应格式错误	content字段格式不对	确保返回`{ content: [{ type: 'text', text: '...' }] }`
stdout污染	console.log干扰	用`process.stderr.write()`替代console
中文乱码	编码问题	确保文件UTF-8编码，JSON.stringify处理

// 不要这样做（已弃用）：
// server.setRequestHandler(SamplingRequestSchema, ...)

// 应该这样做：Server端自行记录日志
import pino from 'pino';

const logger = pino({
  level: process.env.LOG_LEVEL || 'info',
  transport: {
    target: 'pino/file',
    options: { destination: '/tmp/mcp-taskmanager.log' }
  }
});

// 在Tool handler中使用
server.tool('task_create', ..., async (input) => {
  logger.info({ tool: 'task_create', input }, 'Creating task');
  try {
    const task = store.create(input);
    logger.info({ taskId: task.id }, 'Task created successfully');
    return { content: [{ type: 'text', text: JSON.stringify({ success: true, task }) }] };
  } catch (error) {
    logger.error({ error, input }, 'Failed to create task');
    throw error;
  }
});

五、安全加固：生产级防护

5.1 输入验证中间件

import { z, ZodError } from 'zod';

// 通用的安全验证规则
const safeString = (maxLen: number) =>
  z.string().max(maxLen).transform(s => {
    // 防止注入攻击：剥离控制字符
    return s.replace(/[\x00-\x1F\x7F]/g, '').trim();
  });

// 验证工具输入，防止prompt injection
export function sanitizeToolInput(input: Record<string, unknown>): Record<string, unknown> {
  const sanitized: Record<string, unknown> = {};
  for (const [key, value] of Object.entries(input)) {
    if (typeof value === 'string') {
      // 检测潜在的prompt injection模式
      const suspicious = /(<system>|<\/system>|\[INST\]|\[\/INST\]|ignore previous|ignore above)/i;
      if (suspicious.test(value)) {
        sanitized[key] = value.replace(suspicious, '[FILTERED]');
      } else {
        sanitized[key] = value;
      }
    } else {
      sanitized[key] = value;
    }
  }
  return sanitized;
}

5.2 速率限制

export class RateLimiter {
  private calls = new Map<string, number[]>();
  private readonly limit: number;
  private readonly windowMs: number;

  constructor(limit = 60, windowMs = 60000) {
    this.limit = limit;
    this.windowMs = windowMs;
  }

  check(identifier: string): boolean {
    const now = Date.now();
    const windowStart = now - this.windowMs;
    const calls = (this.calls.get(identifier) || [])
      .filter(t => t > windowStart);

    if (calls.length >= this.limit) {
      return false; // 速率限制触发
    }

    calls.push(now);
    this.calls.set(identifier, calls);
    return true;
  }
}

const limiter = new RateLimiter(100, 60000); // 每分钟100次

server.tool('task_create', ..., async (input, extra) => {
  // extra包含请求元信息
  const clientId = extra?.sessionId || 'default';
  if (!limiter.check(clientId)) {
    throw new McpTaskError(
      'Rate limit exceeded. Please try again later.',
      ErrorCodes.RATE_LIMITED,
      429
    );
  }
  // ... 正常逻辑
});

5.3 OAuth 2.1认证（远程Server必需）

如果你的Server以Streamable HTTP方式远程部署，OAuth 2.1 + PKCE是推荐的认证方式：

// src/middleware/auth.ts — 仅用于HTTP传输
import { randomBytes, createHash } from 'crypto';

interface OAuthConfig {
  issuer: string;
  audience: string;
  jwksUri: string;
}

export class OAuthGuard {
  constructor(private config: OAuthConfig) {}

  async validateToken(token: string): Promise<{ sub: string; scope: string }> {
    // 1. 验证JWT签名（使用JWKS）
    // 2. 检查issuer和audience
    // 3. 检查过期时间
    // 4. 返回用户标识和权限范围
    // 实际实现建议使用 jose 库
    throw new Error('Implement with your OAuth provider');
  }

  // PKCE challenge生成
  generateCodeVerifier(): string {
    return randomBytes(32).toString('base64url');
  }

  generateCodeChallenge(verifier: string): string {
    return createHash('sha256')
      .update(verifier)
      .digest('base64url');
  }
}

5.4 安全检查清单

□ 输入验证：所有Tool参数使用Zod schema，设置max/min限制
□ 注入防护：过滤用户输入中的控制字符和prompt injection模式
□ 速率限制：每个Client/Session有调用频率限制
□ 认证（远程）：Streamable HTTP模式启用OAuth 2.1
□ HTTPS：远程部署使用TLS加密
□ 错误信息安全：不泄露内部堆栈、路径、凭证
□ 最小权限：Tool只暴露必要的操作，不暴露文件系统全部访问权
□ 审计日志：记录所有Tool调用和敏感操作
□ 依赖审计：定期运行 npm audit
□ 无硬编码凭证：使用环境变量或密钥管理服务

六、Streamable HTTP传输：远程部署

6.1 为什么从SSE迁移到Streamable HTTP

6.2 实现Streamable HTTP传输

维度	SSE（已弃用）	Streamable HTTP
连接模型	持久连接	可无状态
负载均衡	困难（粘性会话）	原生支持
水平扩展	受限	轻松
重连复杂度	高	低
协议	双通道（HTTP+SSE）	单通道（HTTP POST）

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

async function main() {
  const port = parseInt(process.env.PORT || '3000', 10);

  const httpServer = http.createServer(async (req, res) => {
    // CORS头
    res.setHeader('Access-Control-Allow-Origin', '*');
    res.setHeader('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS');
    res.setHeader('Access-Control-Allow-Headers', 'Content-Type, Authorization');

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

    // MCP endpoint
    if (req.url?.startsWith('/mcp')) {
      const { server } = createServer();
      const sessionId = randomUUID();
      const transport = new StreamableHTTPServerTransport({
        sessionIdGenerator: () => sessionId,
      });

      try {
        await server.connect(transport);
        await transport.handleRequest(req, res);
      } catch (error) {
        res.writeHead(500);
        res.end(JSON.stringify({ error: 'Internal server error' }));
      }
      return;
    }

    // 健康检查
    if (req.url === '/health') {
      res.writeHead(200, { 'Content-Type': 'application/json' });
      res.end(JSON.stringify({ status: 'ok', version: '1.0.0' }));
      return;
    }

    res.writeHead(404);
    res.end();
  });

  httpServer.listen(port, () => {
    process.stderr.write(`MCP Server (Streamable HTTP) listening on port ${port}\n`);
    process.stderr.write(`MCP endpoint: http://localhost:${port}/mcp\n`);
  });
}

main().catch((error) => {
  process.stderr.write(`Fatal error: ${error.message}\n`);
  process.exit(1);
});

6.3 测试远程Server

# 启动HTTP Server
npm run start:http

# 用curl测试健康检查
curl http://localhost:3000/health

# 用Inspector连接远程Server
npx @modelcontextprotocol/inspector --url http://localhost:3000/mcp

6.4 Docker容器化

FROM node:20-slim

WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production
COPY dist/ ./dist/

EXPOSE 3000

HEALTHCHECK --interval=30s --timeout=3s \
  CMD curl -f http://localhost:3000/health || exit 1

CMD ["node", "dist/transport/http.js"]

# 构建并运行
docker build -t mcp-taskmanager .
docker run -p 3000:3000 -e PORT=3000 mcp-taskmanager

七、配置到Claude Code

7.1 stdio模式（本地开发）

# 方法1：使用claude命令行添加
claude mcp add taskmanager -- node /path/to/mcp-server-taskmanager/dist/index.js

# 方法2：手动编辑配置文件

手动配置~/.claude.json（全局）或项目目录.claude/settings.json（项目级）：

{
  "mcpServers": {
    "taskmanager": {
      "command": "node",
      "args": ["/absolute/path/to/mcp-server-taskmanager/dist/index.js"],
      "env": {
        "LOG_LEVEL": "debug"
      }
    }
  }
}

7.2 Streamable HTTP模式（远程Server）

{
  "mcpServers": {
    "taskmanager-remote": {
      "type": "http",
      "url": "https://your-server.com/mcp",
      "headers": {
        "Authorization": "Bearer ${TASKMANAGER_TOKEN}"
      }
    }
  }
}

7.3 验证Server加载

# 在Claude Code中检查MCP Server状态
claude mcp list

# 预期输出：
# taskmanager: connected (3 tools, 2 resources, 1 prompt)

7.4 在Claude Code中使用

# 直接在对话中触发Tool
> 创建一个高优先级任务：完成MCP Server安全审查

# Claude Code会自动调用 task_create Tool
# Tool: task_create
# Args: { title: "完成MCP Server安全审查", priority: "high", ... }

> 列出所有待办任务
# Claude Code会调用 task_list Tool

> 生成今日任务报告
# Claude Code会调用 daily-summary Prompt

八、测试策略：单元测试 + 集成测试

8.1 单元测试：测试Store逻辑

import { describe, it, expect, beforeEach } from 'vitest';
import { TaskStore } from '../src/store/task-store.js';

describe('TaskStore', () => {
  let store: TaskStore;

  beforeEach(() => {
    store = new TaskStore();
  });

  it('should create a task with valid input', () => {
    const task = store.create({
      title: 'Test task',
      description: 'A test description',
      priority: 'high',
    });

    expect(task.id).toMatch(/^task_\d+_/);
    expect(task.title).toBe('Test task');
    expect(task.status).toBe('pending');
    expect(task.priority).toBe('high');
  });

  it('should default invalid priority to medium', () => {
    const task = store.create({
      title: 'Test',
      description: 'Desc',
      priority: 'invalid',
    });
    expect(task.priority).toBe('medium');
  });

  it('should list tasks sorted by creation time', () => {
    store.create({ title: 'First', description: '', priority: 'low' });
    store.create({ title: 'Second', description: '', priority: 'high' });
    const tasks = store.list();
    expect(tasks[0].title).toBe('Second'); // 更新的排在前面
  });

  it('should throw when completing non-existent task', () => {
    expect(() => store.complete('task_nonexistent'))
      .toThrow('Task not found');
  });

  it('should filter tasks by status', () => {
    const t1 = store.create({ title: 'T1', description: '', priority: 'low' });
    store.create({ title: 'T2', description: '', priority: 'low' });
    store.complete(t1.id);

    const pending = store.list({ status: 'pending' });
    expect(pending.length).toBe(1);
    expect(pending[0].title).toBe('T2');
  });
});

8.2 集成测试：测试MCP协议交互

import { describe, it, expect, beforeEach } from 'vitest';
import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js';

describe('MCP Server Integration', () => {
  let client: Client;
  let transport: StdioClientTransport;

  beforeEach(async () => {
    transport = new StdioClientTransport({
      command: 'node',
      args: ['dist/index.js'],
    });

    client = new Client({
      name: 'test-client',
      version: '1.0.0',
    });

    await client.connect(transport);
  });

  it('should list available tools', async () => {
    const { tools } = await client.listTools();
    expect(tools.length).toBeGreaterThanOrEqual(3);
    const names = tools.map(t => t.name);
    expect(names).toContain('task_create');
    expect(names).toContain('task_list');
    expect(names).toContain('task_complete');
  });

  it('should create and list tasks', async () => {
    // 创建任务
    const createResult = await client.callTool({
      name: 'task_create',
      arguments: {
        title: 'Integration test task',
        description: 'Testing MCP integration',
        priority: 'high',
      },
    });

    const created = JSON.parse(
      (createResult.content as Array<{ text: string }>)[0].text
    );
    expect(created.success).toBe(true);
    const taskId = created.task.id;

    // 列出任务
    const listResult = await client.callTool({
      name: 'task_list',
      arguments: {},
    });

    const listed = JSON.parse(
      (listResult.content as Array<{ text: string }>)[0].text
    );
    expect(listed.total).toBeGreaterThanOrEqual(1);
    expect(listed.tasks.some((t: any) => t.id === taskId)).toBe(true);
  });

  it('should list resources', async () => {
    const { resources } = await client.listResources();
    expect(resources.length).toBeGreaterThanOrEqual(1);
    expect(resources.some(r => r.uri.includes('task://'))).toBe(true);
  });

  it('should list prompts', async () => {
    const { prompts } = await client.listPrompts();
    expect(prompts.length).toBeGreaterThanOrEqual(1);
    expect(prompts.some(p => p.name === 'daily-summary')).toBe(true);
  });
});

npm run build && npm test

九、发布：npm包 + MCP Registry

9.1 发布到npm

# 登录npm
npm login

# 确认包名可用
npm view mcp-server-taskmanager

# 发布
npm publish --access public

# 方式1：npx运行（无需安装）
claude mcp add taskmanager -- npx mcp-server-taskmanager

# 方式2：全局安装后运行
npm install -g mcp-server-taskmanager
claude mcp add taskmanager -- mcp-server-taskmanager

9.2 发布到MCP Registry

MCP Registry是2026年3月上线的官方Server发现平台，类似npm但专门服务MCP生态。

# 安装registry CLI
npm install -g @modelcontextprotocol/registry-cli

# 打包
mcp-registry pack

# 发布
mcp-registry publish --name taskmanager --version 1.0.0

{
  "name": "taskmanager",
  "description": "A task management MCP server for Claude Code",
  "version": "1.0.0",
  "repository": "https://github.com/yourname/mcp-server-taskmanager",
  "tools": [
    { "name": "task_create", "description": "Create a new task" },
    { "name": "task_list", "description": "List tasks with filters" },
    { "name": "task_complete", "description": "Mark a task as completed" }
  ],
  "resources": [
    { "uri": "task://all", "description": "All tasks" },
    { "uri": "task:///{id}", "description": "Task by ID" }
  ],
  "prompts": [
    { "name": "daily-summary", "description": "Daily task summary report" }
  ],
  "transport": ["stdio", "streamable-http"],
  "auth": "none"
}

十、完整工作流全景

flowchart TB
    subgraph "开发阶段"
        A[编写Server代码] --> B[TypeScript编译]
        B --> C[启动Inspector调试]
        C --> D{测试通过?}
        D -->|否| A
        D -->|是| E[编写单元测试]
        E --> F[编写集成测试]
        F --> G{覆盖率达标?}
        G -->|否| E
        G -->|是| H[安全审查]
    end

    subgraph "安全审查"
        H --> I[输入验证检查]
        I --> J[速率限制测试]
        J --> K[注入攻击测试]
        K --> L{安全通过?}
        L -->|否| A
        L -->|是| M[构建Docker镜像]
    end

    subgraph "部署阶段"
        M --> N{部署方式}
        N -->|本地| O[配置stdio到Claude Code]
        N -->|远程| P[部署Streamable HTTP]
        P --> Q[配置OAuth认证]
        Q --> R[配置到Claude Code]
        O --> S[端到端验证]
        R --> S
    end

    subgraph "发布阶段"
        S --> T[npm publish]
        T --> U[MCP Registry发布]
        U --> V[编写使用文档]
    end

    style A fill:#e1f5fe
    style C fill:#fff3e0
    style H fill:#fce4ec
    style S fill:#e8f5e9
    style U fill:#f3e5f5

模块	文件数	功能
Tools	3	创建/列表/完成任务
Resources	1	全量任务 + 按ID查询
Prompts	1	每日任务摘要模板
传输层	2	stdio（本地）+ Streamable HTTP（远程）
安全	2	输入验证 + 速率限制
测试	2	单元测试 + MCP协议集成测试
部署	1	Dockerfile

MCP Server工程化：调试安全到生产部署全流程

概述