MCP Server最新规范全实战：结构化输出到异步任务

timeline
    title MCP协议演进时间线
    2024-11 : 初始版本发布\nstdio + SSE双传输\nTools/Resources/Prompts三原语
    2025-03 : 2025-03-26版本\nStreamable HTTP传输替代SSE\n协议基础架构定型
    2025-06 : 2025-06-18版本\n结构化工具输出\nTool Annotations\nElicitation\nResource订阅
    2025-11 : 2025-11-25版本\n异步Tasks\n增强OAuth/OIDC\n协议扩展机制

mkdir modern-mcp-server && cd modern-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zody
npm install -D typescript @types/node tsx

npx tsc --init

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

modern-mcp-server/
├── src/
│   ├── index.ts           # 服务器入口
│   ├── tools/
│   │   ├── query-db.ts    # 结构化输出示例
│   │   ├── deploy.ts      # Tool Annotations示例
│   │   └── migrate.ts     # Elicitation示例
│   ├── resources/
│   │   └── templates.ts   # 资源模板+补全
│   └── utils/
│       └── db.ts          # 模拟数据库
├── package.json
└── tsconfig.json

sequenceDiagram
    participant Client as MCP客户端
    participant Server as MCP服务器

    Client->>Server: tools/call (query_users)
    Note over Server: 执行查询逻辑
    Server->>Client: structuredContent<br/>{users: [...], total: 42}
    Note over Client: 按schema解析<br/>无需文本解析
    Server->>Client: content (文本回退)<br/>"Found 42 users"

// src/utils/db.ts
export interface User {
  id: number;
  name: string;
  email: string;
  role: "admin" | "editor" | "viewer";
  created_at: string;
}

const mockUsers: User[] = [
  { id: 1, name: "Alice Chen", email: "[email protected]", role: "admin", created_at: "2024-01-15" },
  { id: 2, name: "Bob Wang", email: "[email protected]", role: "editor", created_at: "2024-03-22" },
  { id: 3, name: "Carol Li", email: "[email protected]", role: "viewer", created_at: "2024-06-10" },
  { id: 4, name: "David Zhang", email: "[email protected]", role: "editor", created_at: "2024-08-01" },
  { id: 5, name: "Eva Liu", email: "[email protected]", role: "admin", created_at: "2025-01-20" },
];

export function queryUsers(role?: string, limit = 10): { users: User[]; total: number } {
  let filtered = mockUsers;
  if (role) {
    filtered = filtered.filter(u => u.role === role);
  }
  return {
    users: filtered.slice(0, limit),
    total: filtered.length,
  };
}

// src/tools/query-db.ts
import { z } from "zod";
import { queryUsers } from "../utils/db.js";

const QueryUsersSchema = z.object({
  role: z.enum(["admin", "editor", "viewer"]).optional()
    .describe("Filter by user role"),
  limit: z.number().min(1).max(100).default(10)
    .describe("Maximum number of users to return"),
});

// 定义输出schema —— 这是结构化输出的关键
const outputSchema = {
  type: "object" as const,
  properties: {
    users: {
      type: "array" as const,
      items: {
        type: "object" as const,
        properties: {
          id: { type: "number" },
          name: { type: "string" },
          email: { type: "string" },
          role: { type: "string", enum: ["admin", "editor", "viewer"] },
          created_at: { type: "string" },
        },
        required: ["id", "name", "email", "role", "created_at"],
      },
    },
    total: { type: "number" },
  },
  required: ["users", "total"],
};

export function registerQueryUsersTool(server: any) {
  server.tool(
    "query_users",
    "Query users from the database with optional role filter. Returns structured user data.",
    { role: QueryUsersSchema.shape.role, limit: QueryUsersSchema.shape.limit },
    outputSchema,  // ← 声明输出schema
    async (args: z.infer<typeof QueryUsersSchema>) => {
      const result = queryUsers(args.role, args.limit);

      return {
        // 结构化输出 —— 客户端可以按schema直接解析
        structuredContent: result,
        // 文本回退 —— 不支持structuredContent的客户端仍可使用
        content: [{
          type: "text",
          text: `Found ${result.total} users${args.role ? ` with role '${args.role}'` : ""}.\n` +
            result.users.map(u => `  - ${u.name} (${u.email}) [${u.role}]`).join("\n"),
        }],
      };
    }
  );
}

graph LR
    subgraph "Tool Annotations"
        A[readOnlyHint<br/>是否只读] --> D{客户端决策}
        B[destructiveHint<br/>是否破坏性] --> D
        C[idempotentHint<br/>是否幂等] --> D
        E[openWorldHint<br/>是否访问外部] --> D
        F[title<br/>显示标题] --> G{UI展示}
    end
    D --> |autoApprove| H[自动执行]
    D --> |requireConfirm| I[用户确认]

// src/tools/deploy.ts
import { z } from "zod";

const DeploySchema = z.object({
  service: z.string().describe("Service name to deploy"),
  environment: z.enum(["staging", "production"]).describe("Target environment"),
  version: z.string().describe("Version tag to deploy"),
  force: z.boolean().default(false).describe("Force deploy even with warnings"),
});

export function registerDeployTool(server: any) {
  server.tool(
    "deploy_service",
    "Deploy a service to the specified environment",
    {
      service: DeploySchema.shape.service,
      environment: DeploySchema.shape.environment,
      version: DeploySchema.shape.version,
      force: DeploySchema.shape.force,
    },
    {
      // 关键注解声明
      title: "Deploy Service",                        // UI显示标题
      readOnlyHint: false,                            // 会修改状态
      destructiveHint: true,                          // 可能覆盖现有部署
      idempotentHint: true,                           // 部署同一版本幂等
      openWorldHint: true,                            // 访问部署系统
    },
    async (args: z.infer<typeof DeploySchema>) => {
      // 模拟部署逻辑
      const deployId = `deploy-${Date.now()}`;
      const warning = args.environment === "production" && !args.force
        ? "⚠️ Production deploy requires force flag"
        : "";

      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            status: "deployed",
            deploy_id: deployId,
            service: args.service,
            environment: args.environment,
            version: args.version,
            timestamp: new Date().toISOString(),
            warning,
          }, null, 2),
        }],
      };
    }
  );

  // 注册一个只读的部署状态查询工具
  server.tool(
    "check_deploy_status",
    "Check the deployment status of a service",
    {
      service: z.string().describe("Service name"),
      environment: z.enum(["staging", "production"]).describe("Environment"),
    },
    {
      title: "Check Deploy Status",
      readOnlyHint: true,       // 只读，客户端可自动批准
      destructiveHint: false,
      idempotentHint: true,
      openWorldHint: true,      // 需要访问部署系统查询
    },
    async (args) => {
      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            service: args.service,
            environment: args.environment,
            status: "running",
            version: "v2.3.1",
            uptime: "3d 14h 22m",
            last_deploy: "2025-05-20T08:30:00Z",
          }, null, 2),
        }],
      };
    }
  );
}

readOnlyHint: true + destructiveHint: false
  → 自动批准，无需用户确认

destructiveHint: true
  → 强制要求用户确认

openWorldHint: true
  → 沙箱模式下可能被阻止

{
  "permissions": {
    "allow": [
      "mcp__modern-mcp__check_deploy_status"
    ],
    "deny": [
      "mcp__modern-mcp__deploy_service"
    ]
  }
}

sequenceDiagram
    participant Client as MCP客户端<br/>(Claude Code)
    participant Server as MCP服务器
    participant User as 用户

    Client->>Server: tools/call (migrate_database)
    Note over Server: 检测到破坏性操作
    Server->>Client: elicitation/request<br/>"确认要删除旧表吗？"<br/>action: confirm
    Client->>User: 展示确认对话框
    User->>Client: 确认/取消
    Client->>Server: elicitation/response<br/>{action: "accept", content: {confirmed: true}}
    Note over Server: 继续执行迁移
    Server->>Client: 工具结果

// src/tools/migrate.ts
import { z } from "zod";

const MigrateSchema = z.object({
  migration_file: z.string().describe("Path to the migration SQL file"),
  target_db: z.string().describe("Target database name"),
  dry_run: z.boolean().default(true).describe("Run in dry-run mode first"),
});

export function registerMigrateTool(server: any) {
  server.tool(
    "migrate_database",
    "Execute a database migration. Will request user confirmation before destructive operations.",
    {
      migration_file: MigrateSchema.shape.migration_file,
      target_db: MigrateSchema.shape.target_db,
      dry_run: MigrateSchema.shape.dry_run,
    },
    {
      title: "Database Migration",
      readOnlyHint: false,
      destructiveHint: true,
      idempotentHint: false,
      openWorldHint: true,
    },
    async (args: z.infer<typeof MigrateSchema>, extra: any) => {
      // 第一步：Dry run检查
      const dryRunResult = analyzeMigration(args.migration_file);

      if (args.dry_run) {
        return {
          content: [{
            type: "text",
            text: `Dry run result:\n${dryRunResult.summary}\n\nSet dry_run=false to execute for real.`,
          }],
        };
      }

      // 第二步：如果有破坏性操作，通过Elicitation请求用户确认
      if (dryRunResult.hasDestructiveOps) {
        const confirmation = await extra.elicitInput({
          message: `Migration "${args.migration_file}" contains destructive operations:\n` +
            dryRunResult.destructiveOps.map(op => `  • ${op}`).join("\n") +
            `\n\nTarget: ${args.target_db}`,
          requestedSchema: {
            type: "object",
            properties: {
              confirmed: {
                type: "boolean",
                title: "I understand and accept the risks",
              },
              backup_first: {
                type: "boolean",
                title: "Create backup before migration",
                default: true,
              },
            },
            required: ["confirmed"],
          },
        });

        // 用户拒绝或未确认
        if (confirmation.action !== "accept" || !confirmation.content?.confirmed) {
          return {
            content: [{
              type: "text",
              text: "Migration cancelled by user.",
            }],
            isError: true,
          };
        }

        // 用户要求先备份
        if (confirmation.content.backup_first) {
          await performBackup(args.target_db);
        }
      }

      // 第三步：执行迁移
      const result = await executeMigration(args.migration_file, args.target_db);
      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            status: "success",
            migration: args.migration_file,
            database: args.target_db,
            applied_at: new Date().toISOString(),
            changes: result.changes,
          }, null, 2),
        }],
      };
    }
  );
}

// 辅助函数（模拟实现）
function analyzeMigration(file: string) {
  return {
    summary: "3 tables will be modified, 1 index will be dropped",
    hasDestructiveOps: true,
    destructiveOps: [
      "DROP INDEX idx_users_email",
      "ALTER TABLE orders DROP COLUMN legacy_id",
      "TRUNCATE TABLE temp_cache",
    ],
  };
}

async function performBackup(db: string) {
  // 模拟备份
  return { backup_file: `backup_${db}_${Date.now()}.sql` };
}

async function executeMigration(file: string, db: string) {
  return {
    changes: [
      "Applied: create_users_v2_table",
      "Applied: migrate_users_data",
      "Applied: drop_users_v1_table",
    ],
  };
}

db://{database}/tables/{table}
users://{user_id}/profile
logs://{service}/{date}

// src/resources/templates.ts
import { z } from "zod";

export function registerResourceTemplates(server: any) {
  // 资源模板：获取数据库表结构
  server.resource(
    "database-schema",                                    // 资源名称
    "db://{database}/tables/{table}/schema",             // URI模板
    {
      description: "Get the schema of a specific database table",
      mimeType: "application/json",
    },
    async (uri: URL, variables: { database: string; table: string }) => {
      // 根据变量动态生成资源内容
      const schema = getTableSchema(variables.database, variables.table);
      return {
        contents: [{
          uri: uri.href,
          mimeType: "application/json",
          text: JSON.stringify(schema, null, 2),
        }],
      };
    }
  );

  // 资源模板：用户配置
  server.resource(
    "user-profile",
    "users://{user_id}/profile",
    {
      description: "Get user profile information",
      mimeType: "application/json",
    },
    async (uri: URL, variables: { user_id: string }) => {
      const profile = getUserProfile(variables.user_id);
      return {
        contents: [{
          uri: uri.href,
          mimeType: "application/json",
          text: JSON.stringify(profile, null, 2),
        }],
      };
    }
  );

  // 注册Completions —— 为模板参数提供自动补全
  server.completion("database", async (context: any) => {
    // 根据上下文返回可用的数据库名列表
    return {
      values: ["production_db", "staging_db", "analytics_db"],
      total: 3,
      hasMore: false,
    };
  });

  server.completion("table", async (context: any) => {
    // 根据已选的database参数，返回该数据库的表列表
    const database = context?.arguments?.database;
    const tables = getTablesForDatabase(database || "");
    return {
      values: tables,
      total: tables.length,
      hasMore: false,
    };
  });

  server.completion("user_id", async (context: any) => {
    // 返回用户ID列表
    return {
      values: ["1", "2", "3", "4", "5"],
      total: 5,
      hasMore: false,
    };
  });
}

// 模拟数据函数
function getTableSchema(database: string, table: string) {
  const schemas: Record<string, Record<string, any>> = {
    production_db: {
      users: {
        columns: [
          { name: "id", type: "INTEGER", nullable: false, primary_key: true },
          { name: "name", type: "VARCHAR(255)", nullable: false },
          { name: "email", type: "VARCHAR(255)", nullable: false, unique: true },
          { name: "role", type: "ENUM('admin','editor','viewer')", nullable: false },
          { name: "created_at", type: "TIMESTAMP", nullable: false },
        ],
        indexes: ["idx_users_email", "idx_users_role"],
        row_count: 15420,
      },
      orders: {
        columns: [
          { name: "id", type: "INTEGER", nullable: false, primary_key: true },
          { name: "user_id", type: "INTEGER", nullable: false, foreign_key: "users.id" },
          { name: "amount", type: "DECIMAL(10,2)", nullable: false },
          { name: "status", type: "ENUM('pending','completed','cancelled')", nullable: false },
          { name: "created_at", type: "TIMESTAMP", nullable: false },
        ],
        indexes: ["idx_orders_user_id", "idx_orders_status"],
        row_count: 89230,
      },
    },
  };
  return schemas[database]?.[table] || { error: "Table not found" };
}

function getTablesForDatabase(database: string): string[] {
  const dbTables: Record<string, string[]> = {
    production_db: ["users", "orders", "products", "payments"],
    staging_db: ["users", "orders", "test_data"],
    analytics_db: ["events", "page_views", "conversions", "sessions"],
  };
  return dbTables[database] || [];
}

function getUserProfile(userId: string) {
  const users: Record<string, any> = {
    "1": { id: 1, name: "Alice Chen", role: "admin", team: "Platform" },
    "2": { id: 2, name: "Bob Wang", role: "editor", team: "Content" },
    "3": { id: 3, name: "Carol Li", role: "viewer", team: "Marketing" },
  };
  return users[userId] || { error: "User not found" };
}

sequenceDiagram
    participant Client as MCP客户端
    participant Server as MCP服务器

    Client->>Server: resources/templates/list
    Server->>Client: [db://{database}/tables/{table}/schema, ...]

    Client->>Server: completion/complete<br/>ref: {uri_template, name: "database"}
    Server->>Client: {values: ["production_db", "staging_db"]}

    Note over Client: 用户选择 "production_db"

    Client->>Server: completion/complete<br/>ref: {...}, context: {database: "production_db"}<br/>name: "table"
    Server->>Client: {values: ["users", "orders", "products"]}

    Note over Client: 用户选择 "users"

    Client->>Server: resources/read<br/>uri: db://production_db/tables/users/schema
    Server->>Client: {contents: [{text: "{columns: [...]}"}]}

stateDiagram-v2
    [*] --> Submitted: 客户端调用工具
    Submitted --> Running: 服务器开始执行
    Running --> Running: 发送进度通知
    Running --> Completed: 执行成功
    Running --> Failed: 执行失败
    Running --> Cancelled: 客户端取消
    Completed --> [*]
    Failed --> [*]
    Cancelled --> [*]

// src/tools/batch-process.ts
import { z } from "zod";

const BatchProcessSchema = z.object({
  operation: z.enum(["export", "reindex", "cleanup"]).describe("Batch operation type"),
  target: z.string().describe("Target resource identifier"),
  options: z.record(z.any()).optional().describe("Operation-specific options"),
});

export function registerBatchProcessTool(server: any) {
  server.tool(
    "batch_process",
    "Execute a long-running batch operation with progress tracking",
    {
      operation: BatchProcessSchema.shape.operation,
      target: BatchProcessSchema.shape.target,
      options: BatchProcessSchema.shape.options,
    },
    {
      title: "Batch Process",
      readOnlyHint: false,
      destructiveHint: false,
      idempotentHint: false,
      openWorldHint: true,
    },
    async (args: z.infer<typeof BatchProcessSchema>, extra: any) => {
      const totalSteps = 100;
      const progressToken = extra._meta?.progressToken;

      // 模拟长时间运行操作，定期发送进度通知
      for (let step = 0; step <= totalSteps; step++) {
        // 检查取消信号
        if (extra.signal?.aborted) {
          return {
            content: [{
              type: "text",
              text: JSON.stringify({
                status: "cancelled",
                operation: args.operation,
                completed_steps: step,
                total_steps: totalSteps,
              }, null, 2),
            }],
            isError: true,
          };
        }

        // 发送进度通知
        if (progressToken && extra.sendNotification) {
          await extra.sendNotification({
            method: "notifications/progress",
            params: {
              progressToken,
              progress: step,
              total: totalSteps,
            },
          });
        }

        // 模拟实际工作
        await new Promise(resolve => setTimeout(resolve, 100));
      }

      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            status: "completed",
            operation: args.operation,
            target: args.target,
            steps_completed: totalSteps,
            duration_seconds: 10,
            result: {
              processed: 15420,
              succeeded: 15398,
              failed: 22,
            },
          }, null, 2),
        }],
      };
    }
  );
}

// src/index.ts
#!/usr/bin/env node

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

// 创建MCP服务器实例
const server = new McpServer({
  name: "modern-mcp-server",
  version: "1.0.0",
}, {
  capabilities: {
    tools: { listChanged: true },
    resources: { subscribe: true, listChanged: true },
    prompts: { listChanged: true },
    logging: {},
  },
});

// 注册所有工具
import { registerQueryUsersTool } from "./tools/query-db.js";
import { registerDeployTool } from "./tools/deploy.js";
import { registerMigrateTool } from "./tools/migrate.js";
import { registerBatchProcessTool } from "./tools/batch-process.js";
import { registerResourceTemplates } from "./resources/templates.js";

registerQueryUsersTool(server);
registerDeployTool(server);
registerMigrateTool(server);
registerBatchProcessTool(server);
registerResourceTemplates(server);

// 注册一个Prompt示例
server.prompt(
  "analyze_database",
  "Generate a database analysis prompt with current state",
  {
    database: z.string().describe("Database name to analyze"),
    focus: z.enum(["performance", "schema", "data_quality"]).optional()
      .describe("Analysis focus area"),
  },
  async (args) => {
    return {
      messages: [{
        role: "user",
        content: {
          type: "text",
          text: `Please analyze the database "${args.database}" with focus on ${args.focus || "general health"}.\n\n` +
            `Steps:\n` +
            `1. First, use the query_users tool to check user data\n` +
            `2. Then read the resource db://${args.database}/tables/users/schema\n` +
            `3. Provide recommendations based on findings`,
        },
      }],
    };
  }
);

// 启动服务器
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Modern MCP Server running on stdio");
}

main().catch(console.error);

{
  "type": "module",
  "scripts": {
    "build": "tsc",
    "dev": "tsx src/index.ts",
    "start": "node dist/index.js",
    "inspect": "npx @modelcontextprotocol/inspector tsx src/index.ts"
  }
}

npm run inspect

{
  "mcpServers": {
    "modern-mcp": {
      "command": "tsx",
      "args": ["src/index.ts"],
      "cwd": "/path/to/modern-mcp-server"
    }
  }
}

> /mcp

> 运行数据库迁移 tools/migration_001.sql 到 production_db，不用dry run模式

graph TB
    subgraph "MCP客户端 (Claude Code)"
        A[Claude Code CLI]
    end

    subgraph "Modern MCP Server"
        B[McpServer Core]

        subgraph "工具层 Tools"
            C1[query_users<br/>结构化输出]
            C2[deploy_service<br/>Tool Annotations]
            C3[check_deploy_status<br/>只读Annotation]
            C4[migrate_database<br/>Elicitation]
            C5[batch_process<br/>进度通知]
        end

        subgraph "资源层 Resources"
            D1[db://.../schema<br/>资源模板]
            D2[users://.../profile<br/>资源模板]
        end

        subgraph "补全层 Completions"
            E1[database补全]
            E2[table补全]
            E3[user_id补全]
        end

        subgraph "Prompt层"
            F1[analyze_database<br/>结构化Prompt]
        end
    end

    A -->|stdio| B
    B --> C1 & C2 & C3 & C4 & C5
    B --> D1 & D2
    B --> E1 & E2 & E3
    B --> F1

# 构建
npm run build

# 开发模式运行
npm run dev

# Inspector调试
npm run inspect

# 生产运行
npm start

查询数据库中有多少admin角色的用户

检查production环境中web-api服务的部署状态

将web-api v3.0.0部署到production环境

读取production_db中users表的schema

对production_db执行cleanup批处理操作

graph TD
    Start[构建MCP工具] --> Q1{工具是否返回结构化数据?}
    Q1 -->|是| F1[使用 Structured Output<br/>定义outputSchema]
    Q1 -->|否| Q2{工具有副作用吗?}

    F1 --> Q2

    Q2 -->|只读| A1[readOnlyHint: true<br/>destructiveHint: false]
    Q2 -->|有修改| A2[destructiveHint: true/false<br/>根据实际情况]

    A1 & A2 --> Q3{执行中需要用户决策?}
    Q3 -->|是| F3[使用 Elicitation<br/>动态请求确认]
    Q3 -->|否| Q4{执行超过30秒?}
    Q4 -->|是| F5[实现进度通知<br/>考虑Async Tasks]
    Q4 -->|否| Done[完成]

    F3 --> Done
    F5 --> Done

    style F1 fill:#4CAF50,color:white
    style F3 fill:#FF9800,color:white
    style F5 fill:#2196F3,color:white

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

// ... 同样的注册逻辑 ...

const httpServer = createServer(async (req, res) => {
  if (req.method === "POST") {
    const transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: randomUUID,
    });
    await server.connect(transport);
    await transport.handleRequest(req, res);
  }
});

httpServer.listen(3000, () => {
  console.error("MCP Server listening on http://localhost:3000");
});

{
  "mcpServers": {
    "modern-mcp-remote": {
      "url": "http://localhost:3000/mcp",
      "transport": "streamable-http"
    }
  }
}

server.server.setRequestHandler("ping", async () => {
  console.error("[DEBUG] Ping received");
  return {};
});