第4章：MCP Server开发入门

本章将带你编写第一个完整的MCP Server。我们会从零开始，学习Server的基本结构，了解Handlers的工作原理，并通过一个实际的"问候服务器"项目来巩固所学知识。

4.1 Server基本结构

一个完整的MCP Server通常包含以下几个核心部分：

4.1.1 核心组件图解

┌─────────────────────────────────────────────────────────────┐
│                      MCP Server                              │
├─────────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐   │
│  │   Server     │  │ Capabilities │  │    Transport     │   │
│  │   实例       │  │   能力声明    │  │    传输层        │   │
│  └──────────────┘  └──────────────┘  └──────────────────┘   │
├─────────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐   │
│  │   Handlers   │  │   错误处理    │  │    日志记录       │   │
│  │   处理器     │  │              │  │                  │   │
│  └──────────────┘  └──────────────┘  └──────────────────┘   │
└─────────────────────────────────────────────────────────────┘

4.1.2 四步创建Server

无论使用TypeScript还是Python，创建MCP Server都遵循相同的四个步骤：

4.2 TypeScript完整Hello World

让我们从最简单的Hello World开始。这个Server什么都不做，只是证明它可以被正确加载。

4.2.1 完整代码

// server.ts
// MCP Server - Hello World示例 (TypeScript)

// ============================================
// 第1步：导入必要的模块
// ============================================

// 导入Server类，这是MCP Server的核心
import { Server } from '@modelcontextprotocol/sdk/server/index.js';

// 导入StdioServerTransport，使用标准输入输出进行通信
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';

// ============================================
// 第2步：创建Server实例
// ============================================

/**
 * 创建Server实例
 * 第一个参数：Server的元信息（名称和版本）
 * 第二个参数：Server的能力声明（Capabilities）
 */
const server = new Server(
  {
    name: 'hello-server',        // Server的名称，用于标识
    version: '1.0.0'             // Server的版本号
  },
  {
    capabilities: {
      // 声明Server支持的能力
      // 这里我们先声明支持tools（工具）
      tools: {}
    }
  }
);

// ============================================
// 第3步：创建Transport并连接
// ============================================

/**
 * 创建标准输入输出传输层
 * stdio是MCP协议默认的传输方式，通过stdin/stdout通信
 */
const transport = new StdioServerTransport();

/**
 * 将Server连接到Transport
 * 这会启动Server并开始监听连接
 */
await server.connect(transport);

// ============================================
// 启动完成提示（输出到stderr，避免干扰通信）
// ============================================
console.error('Hello World MCP Server 已启动！');
console.error('等待Claude Desktop连接...');

4.2.2 代码详解

4.2.3 运行Server

# 编译TypeScript
npx tsc server.ts --esModuleInterop --target es2022 --module nodenext --moduleResolution nodenext

# 运行Server
node server.js

4.3 Python完整Hello World

Python版本的Hello World同样简单直观。由于Python的语法特性，代码看起来更加简洁。

4.3.1 完整代码

# server.py
# MCP Server - Hello World示例 (Python)

# ============================================
# 第1步：导入必要的模块
# ============================================

import asyncio  # Python异步编程模块

# 导入MCP Server核心类
from mcp.server import Server

# 导入stdio传输层
from mcp.server.stdio import stdio_server

# ============================================
# 第2步：创建Server实例
# ============================================

# 创建Server实例，传入名称
app = Server("hello-server")

# ============================================
# 第3步：定义主函数并启动Server
# ============================================

async def main():
    """
    主函数：启动MCP Server
    使用stdio作为传输层
    """
    # 使用stdio_server作为异步上下文管理器
    # 它会创建read_stream和write_stream用于通信
    async with stdio_server() as (read_stream, write_stream):
        # 运行Server，传入初始化选项
        await app.run(
            read_stream,                    # 读取流
            write_stream,                   # 写入流
            app.create_initialization_options()  # 初始化选项
        )

# ============================================
# 第4步：启动程序
# ============================================

if __name__ == "__main__":
    # 输出启动信息到stderr（避免干扰stdio通信）
    import sys
    print("Hello World MCP Server 已启动！", file=sys.stderr)
    print("等待Claude Desktop连接...", file=sys.stderr)
    
    # 运行异步主函数
    asyncio.run(main())

4.3.2 代码详解

4.3.3 运行Server

# 确保在虚拟环境中
source venv/bin/activate  # macOS/Linux
# 或
venv\Scripts\activate  # Windows

# 运行Server
python server.py

4.4 Handlers（处理器）

Handlers是MCP Server的核心，负责处理客户端发来的各种请求。最常用的Handlers是ListTools和CallTool。

4.4.1 Handler类型一览

4.4.2 ListTools Handler详解

ListTools用于告诉客户端（Claude）这个Server提供了哪些工具。

TypeScript版本：

// 导入工具处理相关的类型
import { 
  ListToolsRequestSchema, 
  CallToolRequestSchema 
} from '@modelcontextprotocol/sdk/types.js';

// 设置ListTools处理器
server.setRequestHandler(ListToolsRequestSchema, async () => {
  // 返回工具列表
  return {
    tools: [
      {
        name: 'greet',                    // 工具名称
        description: '向指定的人问好',     // 工具描述（AI通过描述了解工具用途）
        inputSchema: {                    // 输入参数模式（JSON Schema格式）
          type: 'object',
          properties: {
            name: {
              type: 'string',
              description: '要问候的人的名字'
            }
          },
          required: ['name']              // 必需参数
        }
      }
    ]
  };
});

Python版本：

# 导入工具相关的类型
from mcp.types import Tool, TextContent

# 使用装饰器注册ListTools处理器
@app.list_tools()
async def list_tools() -> list[Tool]:
    """
    返回Server提供的所有工具列表
    """
    return [
        Tool(
            name="greet",                           # 工具名称
            description="向指定的人问好",            # 工具描述
            inputSchema={                           # 输入参数模式
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "要问候的人的名字"
                    }
                },
                "required": ["name"]
            }
        )
    ]

4.4.3 CallTool Handler详解

CallTool用于实际执行工具的功能。当AI决定使用某个工具时，会触发这个Handler。

TypeScript版本：

// 设置CallTool处理器
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  // request参数包含调用信息
  const { name, arguments: args } = request.params;
  
  // 根据工具名称执行对应逻辑
  if (name === 'greet') {
    const userName = args.name as string;
    
    // 返回工具执行结果
    return {
      content: [
        {
          type: 'text',
          text: `你好，${userName}！欢迎使用MCP！`
        }
      ]
    };
  }
  
  // 未知工具返回错误
  throw new Error(`未知工具: ${name}`);
});

Python版本：

# 使用装饰器注册CallTool处理器
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    """
    执行指定的工具
    
    Args:
        name: 要调用的工具名称
        arguments: 工具调用参数
    
    Returns:
        工具执行结果列表
    """
    if name == "greet":
        user_name = arguments.get("name", "朋友")
        
        # 返回工具执行结果
        return [
            TextContent(
                type="text",
                text=f"你好，{user_name}！欢迎使用MCP！"
            )
        ]
    
    # 未知工具返回错误
    raise ValueError(f"未知工具: {name}")

4.5 错误处理

良好的错误处理是生产级Server的必备要素。MCP定义了标准的错误码和错误格式。

4.5.1 标准错误码

4.5.2 错误返回格式

TypeScript错误处理：

import { McpError, ErrorCode } from '@modelcontextprotocol/sdk/types.js';

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  try {
    const { name, arguments: args } = request.params;
    
    if (name === 'greet') {
      // 参数验证
      if (!args.name || typeof args.name !== 'string') {
        // 使用标准错误码
        throw new McpError(
          ErrorCode.InvalidParams,
          '缺少必需的参数: name (string类型)'
        );
      }
      
      return {
        content: [{
          type: 'text',
          text: `你好，${args.name}！`
        }]
      };
    }
    
    // 工具不存在
    throw new McpError(
      ErrorCode.MethodNotFound,
      `工具 '${name}' 不存在`
    );
    
  } catch (error) {
    // 已经是McpError则直接抛出
    if (error instanceof McpError) {
      throw error;
    }
    
    // 其他错误包装为InternalError
    console.error('执行工具时出错:', error);
    throw new McpError(
      ErrorCode.InternalError,
      `执行失败: ${error instanceof Error ? error.message : String(error)}`
    );
  }
});

Python错误处理：

from mcp.types import McpError
from mcp.types import ErrorCode

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    try:
        if name == "greet":
            # 参数验证
            user_name = arguments.get("name")
            if not user_name or not isinstance(user_name, str):
                raise McpError(
                    ErrorCode.InvalidParams,
                    "缺少必需的参数: name (string类型)"
                )
            
            return [
                TextContent(
                    type="text",
                    text=f"你好，{user_name}！"
                )
            ]
        
        # 工具不存在
        raise McpError(
            ErrorCode.MethodNotFound,
            f"工具 '{name}' 不存在"
        )
        
    except McpError:
        # 已经是McpError则直接抛出
        raise
    except Exception as e:
        # 其他错误包装为InternalError
        import traceback
        traceback.print_exc()
        raise McpError(
            ErrorCode.InternalError,
            f"执行失败: {str(e)}"
        )

4.6 日志记录

日志是调试和监控Server运行状态的重要手段。由于MCP使用stdio通信，日志需要输出到stderr。

4.6.1 TypeScript日志

// 简单的日志工具
const logger = {
  debug: (msg: string) => console.error(`[DEBUG] ${msg}`),
  info: (msg: string) => console.error(`[INFO] ${msg}`),
  warn: (msg: string) => console.error(`[WARN] ${msg}`),
  error: (msg: string) => console.error(`[ERROR] ${msg}`)
};

// 使用示例
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  logger.info(`收到工具调用请求: ${request.params.name}`);
  logger.debug(`参数: ${JSON.stringify(request.params.arguments)}`);
  
  try {
    // ... 执行逻辑
    logger.info('工具执行成功');
    return result;
  } catch (error) {
    logger.error(`执行失败: ${error}`);
    throw error;
  }
});

4.6.2 Python日志

import logging
import sys

# 配置日志输出到stderr
logging.basicConfig(
    level=logging.DEBUG,
    format='[%(levelname)s] %(message)s',
    stream=sys.stderr
)

logger = logging.getLogger(__name__)

# 使用示例
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    logger.info(f"收到工具调用请求: {name}")
    logger.debug(f"参数: {arguments}")
    
    try:
        # ... 执行逻辑
        logger.info("工具执行成功")
        return result
    except Exception as e:
        logger.error(f"执行失败: {e}")
        raise

4.7 完整实战：问候服务器

现在，让我们把所有知识点整合起来，开发一个完整的"问候服务器"。这个Server接受名字参数，返回个性化的问候语。

4.7.1 项目结构

greet-server/
├── package.json          # 项目配置（TypeScript版本）
├── tsconfig.json         # TypeScript配置
└── src/
    └── server.ts         # 主程序

或

greet-server/
├── requirements.txt      # Python依赖
└── server.py            # 主程序

4.7.2 TypeScript完整版本

// src/server.ts
// 完整版问候服务器 - TypeScript

// ============================================
// 导入模块
// ============================================
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { 
  ListToolsRequestSchema,
  CallToolRequestSchema,
  McpError,
  ErrorCode
} from '@modelcontextprotocol/sdk/types.js';

// ============================================
// 日志工具
// ============================================
const logger = {
  info: (msg: string) => console.error(`[INFO] ${new Date().toISOString()} ${msg}`),
  debug: (msg: string) => console.error(`[DEBUG] ${new Date().toISOString()} ${msg}`),
  error: (msg: string) => console.error(`[ERROR] ${new Date().toISOString()} ${msg}`)
};

// ============================================
// 创建Server实例
// ============================================
const server = new Server(
  {
    name: 'greet-server',
    version: '1.0.0'
  },
  {
    capabilities: { tools: {} }
  }
);

// ============================================
// ListTools Handler - 列出可用工具
// ============================================
server.setRequestHandler(ListToolsRequestSchema, async () => {
  logger.debug('收到ListTools请求');
  
  return {
    tools: [
      {
        name: 'greet',
        description: '向指定的人发送个性化问候语。支持根据时间自动选择问候语（早上好/下午好/晚上好）',
        inputSchema: {
          type: 'object',
          properties: {
            name: {
              type: 'string',
              description: '要问候的人的名字'
            },
            language: {
              type: 'string',
              enum: ['zh', 'en'],
              description: '问候语语言，可选值：zh（中文）、en（英文），默认为zh',
              default: 'zh'
            }
          },
          required: ['name']
        }
      }
    ]
  };
});

// ============================================
// CallTool Handler - 执行工具
// ============================================
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  
  logger.info(`执行工具: ${name}`);
  logger.debug(`参数: ${JSON.stringify(args)}`);
  
  try {
    if (name === 'greet') {
      // 参数验证
      if (!args.name || typeof args.name !== 'string') {
        throw new McpError(
          ErrorCode.InvalidParams,
          '参数"name"是必需的，且必须是字符串类型'
        );
      }
      
      const userName = args.name as string;
      const language = (args.language as string) || 'zh';
      
      // 根据时间生成问候语
      const hour = new Date().getHours();
      let timeGreeting = '';
      
      if (language === 'zh') {
        if (hour < 12) timeGreeting = '早上好';
        else if (hour < 18) timeGreeting = '下午好';
        else timeGreeting = '晚上好';
      } else {
        if (hour < 12) timeGreeting = 'Good morning';
        else if (hour < 18) timeGreeting = 'Good afternoon';
        else timeGreeting = 'Good evening';
      }
      
      // 生成问候消息
      const message = language === 'zh'
        ? `${timeGreeting}，${userName}！欢迎使用MCP！🎉`
        : `${timeGreeting}, ${userName}! Welcome to MCP! 🎉`;
      
      logger.info('问候工具执行成功');
      
      return {
        content: [
          {
            type: 'text',
            text: message
          }
        ]
      };
    }
    
    // 未知工具
    throw new McpError(
      ErrorCode.MethodNotFound,
      `未找到工具: ${name}。可用工具: greet`
    );
    
  } catch (error) {
    if (error instanceof McpError) {
      throw error;
    }
    
    logger.error(`执行失败: ${error}`);
    throw new McpError(
      ErrorCode.InternalError,
      `服务器内部错误: ${error instanceof Error ? error.message : String(error)}`
    );
  }
});

// ============================================
// 启动Server
// ============================================
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  logger.info('问候服务器已启动，等待连接...');
}

main().catch((error) => {
  logger.error(`启动失败: ${error}`);
  process.exit(1);
});

4.7.3 Python完整版本

# server.py
# 完整版问候服务器 - Python

# ============================================
# 导入模块
# ============================================
import asyncio
import sys
import logging
from datetime import datetime

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, McpError, ErrorCode

# ============================================
# 配置日志
# ============================================
logging.basicConfig(
    level=logging.INFO,
    format='[%(levelname)s] %(asctime)s %(message)s',
    stream=sys.stderr
)
logger = logging.getLogger(__name__)

# ============================================
# 创建Server实例
# ============================================
app = Server("greet-server")

# ============================================
# ListTools Handler - 列出可用工具
# ============================================
@app.list_tools()
async def list_tools() -> list[Tool]:
    """
    返回Server提供的所有工具列表
    """
    logger.debug("收到ListTools请求")
    
    return [
        Tool(
            name="greet",
            description="向指定的人发送个性化问候语。支持根据时间自动选择问候语（早上好/下午好/晚上好）",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "要问候的人的名字"
                    },
                    "language": {
                        "type": "string",
                        "enum": ["zh", "en"],
                        "description": "问候语语言，可选值：zh（中文）、en（英文），默认为zh"
                    }
                },
                "required": ["name"]
            }
        )
    ]

# ============================================
# CallTool Handler - 执行工具
# ============================================
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    """
    执行指定的工具
    
    Args:
        name: 要调用的工具名称
        arguments: 工具调用参数
    
    Returns:
        工具执行结果列表
    """
    logger.info(f"执行工具: {name}")
    logger.debug(f"参数: {arguments}")
    
    try:
        if name == "greet":
            # 参数验证
            user_name = arguments.get("name")
            if not user_name or not isinstance(user_name, str):
                raise McpError(
                    ErrorCode.InvalidParams,
                    '参数"name"是必需的，且必须是字符串类型'
                )
            
            language = arguments.get("language", "zh")
            
            # 根据时间生成问候语
            hour = datetime.now().hour
            time_greeting = ""
            
            if language == "zh":
                if hour < 12:
                    time_greeting = "早上好"
                elif hour < 18:
                    time_greeting = "下午好"
                else:
                    time_greeting = "晚上好"
            else:
                if hour < 12:
                    time_greeting = "Good morning"
                elif hour < 18:
                    time_greeting = "Good afternoon"
                else:
                    time_greeting = "Good evening"
            
            # 生成问候消息
            if language == "zh":
                message = f"{time_greeting}，{user_name}！欢迎使用MCP！🎉"
            else:
                message = f"{time_greeting}, {user_name}! Welcome to MCP! 🎉"
            
            logger.info("问候工具执行成功")
            
            return [
                TextContent(
                    type="text",
                    text=message
                )
            ]
        
        # 未知工具
        raise McpError(
            ErrorCode.MethodNotFound,
            f"未找到工具: {name}。可用工具: greet"
        )
        
    except McpError:
        raise
    except Exception as e:
        logger.error(f"执行失败: {e}")
        import traceback
        traceback.print_exc()
        raise McpError(
            ErrorCode.InternalError,
            f"服务器内部错误: {str(e)}"
        )

# ============================================
# 主函数
# ============================================
async def main():
    """
    主函数：启动MCP Server
    """
    async with stdio_server() as (read_stream, write_stream):
        logger.info("问候服务器已启动，等待连接...")
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

# ============================================
# 启动程序
# ============================================
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("收到中断信号，服务器已关闭")
    except Exception as e:
        logger.error(f"服务器异常: {e}")
        sys.exit(1)

4.7.4 配置Claude Desktop

将Server配置到Claude Desktop中：

{
  "mcpServers": {
    "greet": {
      "command": "node",
      "args": ["C:/projects/greet-server/dist/server.js"]
    }
  }
}

或Python版本：

{
  "mcpServers": {
    "greet": {
      "command": "C:/projects/greet-server/venv/Scripts/python.exe",
      "args": ["C:/projects/greet-server/server.py"]
    }
  }
}

4.7.5 测试Server

配置完成后，在Claude Desktop中测试：

1. 重启Claude Desktop
2. 在对话中输入：请使用greet工具向小明问好
3. Claude会调用greet工具，并返回：早上好，小明！欢迎使用MCP！🎉（根据当前时间）

本章小结

恭喜！你已经完成了第一个完整的MCP Server开发。下一章我们将学习Resources（资源），了解如何让AI访问外部数据。