零基础构建MCP服务器:TypeScript/Python双语言实战指南
🌟 Hello,我是摘星!
🌈 在彩虹般绚烂的技术栈中,我是那个永不停歇的色彩收集者。
🦋 每一个优化都是我培育的花朵,每一个特性都是我放飞的蝴蝶。
🔬 每一次代码审查都是我的显微镜观察,每一次重构都是我的化学实验。
🎵 在编程的交响乐中,我既是指挥家也是演奏者。让我们一起,在技术的音乐厅里,奏响属于程序员的华美乐章。
摘要
作为一名深耕技术领域多年的博主摘星,我深刻感受到了MCP(Model Context Protocol)协议在AI生态系统中的革命性意义。MCP作为Anthropic推出的开放标准,正在重新定义AI应用与外部系统的交互方式,它不仅解决了传统API集成的复杂性问题,更为开发者提供了一个统一、安全、高效的连接框架。在过去几个月的实践中,我发现许多开发者对MCP的概念理解透彻,但在实际动手构建MCP服务器时却遇到了各种技术壁垒。从环境配置的细节问题到SDK API的深度理解,从第一个Hello World程序的调试到生产环境的部署优化,每一个环节都可能成为初学者的绊脚石。因此,我决定撰写这篇全面的实战指南,以零基础开发者的视角,通过TypeScript和Python两种主流语言的并行讲解,帮助大家从理论走向实践。本文将涵盖完整的开发环境搭建流程、MCP SDK核心API的深度剖析、实际项目的逐步构建过程,以及我在开发过程中积累的调试技巧和问题解决方案。我相信,通过这篇文章的学习,读者不仅能够掌握MCP服务器开发的核心技能,更能够理解MCP协议的设计哲学和最佳实践,为后续的深度开发和创新应用奠定坚实基础。
1. MCP协议基础概念
1.1 什么是MCP服务器
MCP(Model Context Protocol)服务器是一个实现了MCP协议标准的应用程序,它作为AI模型与外部资源之间的桥梁,提供工具(Tools)、资源(Resources)和提示(Prompts)等功能。
图1 MCP服务器架构图
1.2 核心组件说明
组件 |
功能描述 |
使用场景 |
Tools |
可执行的函数接口 |
API调用、数据处理、系统操作 |
Resources |
可访问的数据源 |
文件读取、数据库查询、配置获取 |
Prompts |
预定义的提示模板 |
对话引导、任务模板、上下文设置 |
2. 开发环境搭建与配置
2.1 TypeScript开发环境
2.1.1 基础环境准备
# 安装Node.js (推荐v18+) node --version npm --version # 创建项目目录 mkdir mcp-server-ts cd mcp-server-ts # 初始化项目 npm init -y
2.1.2 依赖包安装
# 安装MCP SDK npm install @modelcontextprotocol/sdk # 安装TypeScript相关依赖 npm install -D typescript @types/node ts-node nodemon # 安装其他工具库 npm install zod dotenv
2.1.3 TypeScript配置
// tsconfig.json { "compilerOptions": { "target": "ES2020", "module": "commonjs", "lib": ["ES2020"], "outDir": "./dist", "rootDir": "./src", "strict": true, "esModuleInterop": true, "skipLibCheck": true, "forceConsistentCasingInFileNames": true, "resolveJsonModule": true, "declaration": true, "declarationMap": true, "sourceMap": true }, "include": ["src/**/*"], "exclude": ["node_modules", "dist"] }
2.2 Python开发环境
2.2.1 虚拟环境创建
# 创建项目目录 mkdir mcp-server-py cd mcp-server-py # 创建虚拟环境 python -m venv venv # 激活虚拟环境 (Windows) venv\Scripts\activate # 激活虚拟环境 (Linux/Mac) source venv/bin/activate
2.2.2 依赖包安装
# 安装MCP SDK pip install mcp # 安装开发工具 pip install python-dotenv pydantic typing-extensions # 生成requirements.txt pip freeze > requirements.txt
2.3 开发工具配置
图2 开发环境配置流程图
3. MCP SDK核心API详解
3.1 服务器初始化
3.1.1 TypeScript实现
// src/server.ts import { Server } from '@modelcontextprotocol/sdk/server/index.js'; import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'; import { CallToolRequestSchema, ListToolsRequestSchema, Tool, } from '@modelcontextprotocol/sdk/types.js'; class MCPServer { private server: Server; constructor() { // 初始化服务器实例 this.server = new Server( { name: 'example-server', version: '0.1.0', }, { capabilities: { tools: {}, // 支持工具功能 resources: {}, // 支持资源功能 prompts: {}, // 支持提示功能 }, } ); this.setupHandlers(); } private setupHandlers(): void { // 设置工具列表处理器 this.server.setRequestHandler(ListToolsRequestSchema, async () => { return { tools: this.getAvailableTools(), }; }); // 设置工具调用处理器 this.server.setRequestHandler(CallToolRequestSchema, async (request) => { return await this.handleToolCall(request.params); }); } private getAvailableTools(): Tool[] { return [ { name: 'echo', description: '回显输入的文本', inputSchema: { type: 'object', properties: { text: { type: 'string', description: '要回显的文本', }, }, required: ['text'], }, }, ]; } private async handleToolCall(params: any) { const { name, arguments: args } = params; switch (name) { case 'echo': return { content: [ { type: 'text', text: `Echo: ${args.text}`, }, ], }; default: throw new Error(`Unknown tool: ${name}`); } } async start(): Promise<void> { const transport = new StdioServerTransport(); await this.server.connect(transport); console.error('MCP Server started successfully'); } } // 启动服务器 const server = new MCPServer(); server.start().catch(console.error);
3.1.2 Python实现
# server.py import asyncio import logging from typing import Any, Sequence from mcp.server import Server from mcp.server.models import InitializationOptions from mcp.server.stdio import stdio_server from mcp.types import ( CallToolRequest, ListToolsRequest, Tool, TextContent, CallToolResult, ) # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class MCPServer: def __init__(self): self.server = Server("example-server") self.setup_handlers() def setup_handlers(self): """设置请求处理器""" @self.server.list_tools() async def handle_list_tools() -> list[Tool]: """返回可用工具列表""" return [ Tool( name="echo", description="回显输入的文本", inputSchema={ "type": "object", "properties": { "text": { "type": "string", "description": "要回显的文本", } }, "required": ["text"], }, ) ] @self.server.call_tool() async def handle_call_tool( name: str, arguments: dict[str, Any] ) -> list[TextContent]: """处理工具调用""" if name == "echo": text = arguments.get("text", "") return [ TextContent( type="text", text=f"Echo: {text}", ) ] else: raise ValueError(f"Unknown tool: {name}") async def main(): """主函数""" server = MCPServer() # 使用stdio传输启动服务器 async with stdio_server() as (read_stream, write_stream): await server.server.run( read_stream, write_stream, InitializationOptions( server_name="example-server", server_version="0.1.0", capabilities=server.server.get_capabilities( notification_options=None, experimental_capabilities=None, ), ), ) if __name__ == "__main__": asyncio.run(main())
3.2 核心API对比分析
功能 |
TypeScript |
Python |
特点 |
服务器初始化 |
|
|
类似的构造方式 |
请求处理 |
|
装饰器模式 |
Python更简洁 |
类型安全 |
编译时检查 |
运行时检查 |
TS类型更严格 |
异步处理 |
Promise/async |
asyncio |
都支持现代异步 |
4. 第一个Hello World服务器
4.1 项目结构设计
图3 项目结构组织图
4.2 完整实现示例
4.2.1 TypeScript版本
// src/handlers/tools.ts import { Tool, CallToolRequest } from '@modelcontextprotocol/sdk/types.js'; export class ToolHandler { getTools(): Tool[] { return [ { name: 'hello', description: '返回问候语', inputSchema: { type: 'object', properties: { name: { type: 'string', description: '要问候的名字', default: 'World' } } } }, { name: 'calculate', description: '执行简单的数学计算', inputSchema: { type: 'object', properties: { operation: { type: 'string', enum: ['add', 'subtract', 'multiply', 'divide'], description: '计算操作类型' }, a: { type: 'number', description: '第一个数字' }, b: { type: 'number', description: '第二个数字' } }, required: ['operation', 'a', 'b'] } } ]; } async handleCall(name: string, args: any) { switch (name) { case 'hello': return this.handleHello(args); case 'calculate': return this.handleCalculate(args); default: throw new Error(`未知工具: ${name}`); } } private handleHello(args: any) { const name = args.name || 'World'; return { content: [ { type: 'text', text: `Hello, ${name}! 欢迎使用MCP服务器!` } ] }; } private handleCalculate(args: any) { const { operation, a, b } = args; let result: number; switch (operation) { case 'add': result = a + b; break; case 'subtract': result = a - b; break; case 'multiply': result = a * b; break; case 'divide': if (b === 0) { throw new Error('除数不能为零'); } result = a / b; break; default: throw new Error(`不支持的操作: ${operation}`); } return { content: [ { type: 'text', text: `计算结果: ${a} ${operation} ${b} = ${result}` } ] }; } }
4.2.2 Python版本
# handlers/tools.py from typing import Any, List from mcp.types import Tool, TextContent class ToolHandler: def get_tools(self) -> List[Tool]: """返回可用工具列表""" return [ Tool( name="hello", description="返回问候语", inputSchema={ "type": "object", "properties": { "name": { "type": "string", "description": "要问候的名字", "default": "World" } } } ), Tool( name="calculate", description="执行简单的数学计算", inputSchema={ "type": "object", "properties": { "operation": { "type": "string", "enum": ["add", "subtract", "multiply", "divide"], "description": "计算操作类型" }, "a": { "type": "number", "description": "第一个数字" }, "b": { "type": "number", "description": "第二个数字" } }, "required": ["operation", "a", "b"] } ) ] async def handle_call(self, name: str, arguments: dict[str, Any]) -> List[TextContent]: """处理工具调用""" if name == "hello": return await self._handle_hello(arguments) elif name == "calculate": return await self._handle_calculate(arguments) else: raise ValueError(f"未知工具: {name}") async def _handle_hello(self, args: dict[str, Any]) -> List[TextContent]: """处理问候工具""" name = args.get("name", "World") return [ TextContent( type="text", text=f"Hello, {name}! 欢迎使用MCP服务器!" ) ] async def _handle_calculate(self, args: dict[str, Any]) -> List[TextContent]: """处理计算工具""" operation = args["operation"] a = args["a"] b = args["b"] if operation == "add": result = a + b elif operation == "subtract": result = a - b elif operation == "multiply": result = a * b elif operation == "divide": if b == 0: raise ValueError("除数不能为零") result = a / b else: raise ValueError(f"不支持的操作: {operation}") return [ TextContent( type="text", text=f"计算结果: {a} {operation} {b} = {result}" ) ]
4.3 服务器启动脚本
// package.json (TypeScript) { "name": "mcp-hello-world", "version": "1.0.0", "scripts": { "build": "tsc", "start": "node dist/server.js", "dev": "ts-node src/server.ts", "watch": "nodemon --exec ts-node src/server.ts" } }
# run.py (Python) #!/usr/bin/env python3 import asyncio import sys from server import main if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n服务器已停止") sys.exit(0) except Exception as e: print(f"服务器启动失败: {e}") sys.exit(1)
5. 调试技巧与常见问题解决
5.1 调试工具配置
5.1.1 日志系统设置
// src/utils/logger.ts export class Logger { private static instance: Logger; static getInstance(): Logger { if (!Logger.instance) { Logger.instance = new Logger(); } return Logger.instance; } info(message: string, data?: any): void { console.error(`[INFO] ${new Date().toISOString()}: ${message}`); if (data) { console.error(JSON.stringify(data, null, 2)); } } error(message: string, error?: Error): void { console.error(`[ERROR] ${new Date().toISOString()}: ${message}`); if (error) { console.error(error.stack); } } debug(message: string, data?: any): void { if (process.env.DEBUG === 'true') { console.error(`[DEBUG] ${new Date().toISOString()}: ${message}`); if (data) { console.error(JSON.stringify(data, null, 2)); } } } }
5.1.2 MCP Inspector使用
# 安装MCP Inspector npm install -g @modelcontextprotocol/inspector # 启动调试会话 mcp-inspector typescript ts-node src/server.ts
5.2 常见问题解决方案
图4 常见问题分类图
5.3 问题诊断清单
问题类型 |
症状 |
解决方案 |
预防措施 |
连接失败 |
客户端无法连接 |
检查端口和防火墙 |
使用健康检查 |
协议错误 |
JSON-RPC解析失败 |
验证消息格式 |
使用Schema验证 |
内存泄漏 |
服务器内存持续增长 |
检查事件监听器 |
定期内存监控 |
响应超时 |
工具调用无响应 |
增加超时时间 |
异步处理优化 |
5.4 性能优化建议
"优化不是过早的,而是持续的。在MCP服务器开发中,性能监控应该从第一行代码开始。" —— 性能优化最佳实践
// 性能监控示例 class PerformanceMonitor { private metrics: Map<string, number[]> = new Map(); startTimer(operation: string): () => void { const start = Date.now(); return () => { const duration = Date.now() - start; if (!this.metrics.has(operation)) { this.metrics.set(operation, []); } this.metrics.get(operation)!.push(duration); }; } getStats(operation: string) { const times = this.metrics.get(operation) || []; if (times.length === 0) return null; const avg = times.reduce((a, b) => a + b, 0) / times.length; const max = Math.max(...times); const min = Math.min(...times); return { avg, max, min, count: times.length }; } }
6. 进阶功能实现
6.1 资源管理
// 资源处理器实现 export class ResourceHandler { async listResources() { return [ { uri: 'file://config.json', name: '配置文件', description: '服务器配置信息', mimeType: 'application/json' } ]; } async readResource(uri: string) { if (uri === 'file://config.json') { return { contents: [ { uri, mimeType: 'application/json', text: JSON.stringify({ version: '1.0.0', features: ['tools', 'resources'] }, null, 2) } ] }; } throw new Error(`资源未找到: ${uri}`); } }
6.2 提示模板系统
7. 测试与质量保证
7.1 单元测试框架
// tests/tools.test.ts import { ToolHandler } from '../src/handlers/tools'; describe('ToolHandler', () => { let handler: ToolHandler; beforeEach(() => { handler = new ToolHandler(); }); test('should return greeting', async () => { const result = await handler.handleCall('hello', { name: 'Test' }); expect(result.content[0].text).toContain('Hello, Test!'); }); test('should calculate addition', async () => { const result = await handler.handleCall('calculate', { operation: 'add', a: 5, b: 3 }); expect(result.content[0].text).toContain('= 8'); }); });
提示处理器
class PromptHandler: def get_prompts(self) -> List[Prompt]: return [ Prompt( name="code_review", description="代码审查提示模板", arguments=[ PromptArgument( name="language", description="编程语言", required=True ), PromptArgument( name="code", description="要审查的代码", required=True ) ] ) ] async def get_prompt(self, name: str, arguments: dict) -> GetPromptResult: if name == "code_review": language = arguments.get("language", "") code = arguments.get("code", "") prompt_text = f"""
请审查以下{language}代码:
{code} 请从以下方面进行评估: 1. 代码质量和可读性 2. 性能优化建议 3. 安全性考虑 4. 最佳实践遵循情况 """ return GetPromptResult( description=f"{language}代码审查", messages=[ PromptMessage( role="user", content=TextContent(type="text", text=prompt_text) ) ] ) raise ValueError(f"未知提示: {name}")
7.2 集成测试
# tests/test_integration.py import pytest import asyncio from server import MCPServer @pytest.mark.asyncio async def test_server_initialization(): """测试服务器初始化""" server = MCPServer() assert server.server is not None @pytest.mark.asyncio async def test_tool_execution(): """测试工具执行""" server = MCPServer() tools = await server.tool_handler.get_tools() assert len(tools) > 0 assert any(tool.name == "hello" for tool in tools)
总结
经过这篇详尽的实战指南,我作为博主摘星想要与大家分享一些深度思考和实践心得。MCP服务器开发虽然在技术层面并不复杂,但要构建一个真正生产可用、性能优异的服务器,需要我们在多个维度上精益求精。从我的开发经验来看,TypeScript和Python两种语言各有优势:TypeScript在类型安全和开发体验上表现卓越,特别适合大型项目和团队协作;而Python则在快速原型开发和AI生态集成方面更具优势,其简洁的语法和丰富的库生态让开发者能够快速实现复杂功能。在实际项目中,我建议开发者根据团队技术栈、项目规模和性能要求来选择合适的语言。更重要的是,MCP协议的真正价值不仅在于技术实现,更在于它为AI应用生态带来的标准化和互操作性。通过本文的学习,我们不仅掌握了MCP服务器的开发技能,更重要的是理解了现代AI应用架构的设计思想。随着AI技术的快速发展,MCP协议必将成为连接AI模型与外部世界的重要桥梁,而掌握MCP服务器开发技能的开发者,也将在这个充满机遇的时代中占据先机。我鼓励每一位读者都能够动手实践,从简单的Hello World开始,逐步构建更复杂、更实用的MCP服务器,为AI应用的发展贡献自己的力量。
参考资源
🌈 我是摘星!如果这篇文章在你的技术成长路上留下了印记:
👁️ 【关注】与我一起探索技术的无限可能,见证每一次突破
👍 【点赞】为优质技术内容点亮明灯,传递知识的力量
🔖 【收藏】将精华内容珍藏,随时回顾技术要点
💬 【评论】分享你的独特见解,让思维碰撞出智慧火花
🗳️ 【投票】用你的选择为技术社区贡献一份力量
技术路漫漫,让我们携手前行,在代码的世界里摘取属于程序员的那片星辰大海!