当前位置: 首页 > news >正文

零基础构建MCP服务器:TypeScript/Python双语言实战指南

零基础构建MCP服务器:TypeScript/Python双语言实战指南

🌟 Hello,我是摘星!

🌈 在彩虹般绚烂的技术栈中,我是那个永不停歇的色彩收集者。

🦋 每一个优化都是我培育的花朵,每一个特性都是我放飞的蝴蝶。

🔬 每一次代码审查都是我的显微镜观察,每一次重构都是我的化学实验。

🎵 在编程的交响乐中,我既是指挥家也是演奏者。让我们一起,在技术的音乐厅里,奏响属于程序员的华美乐章。

目录

零基础构建MCP服务器:TypeScript/Python双语言实战指南

摘要

1. MCP协议基础概念

1.1 什么是MCP服务器

1.2 核心组件说明

2. 开发环境搭建与配置

2.1 TypeScript开发环境

2.1.1 基础环境准备

2.1.2 依赖包安装

2.1.3 TypeScript配置

2.2 Python开发环境

2.2.1 虚拟环境创建

2.2.2 依赖包安装

2.3 开发工具配置

3. MCP SDK核心API详解

3.1 服务器初始化

3.1.1 TypeScript实现

3.1.2 Python实现

3.2 核心API对比分析

4. 第一个Hello World服务器

4.1 项目结构设计

4.2 完整实现示例

4.2.1 TypeScript版本

4.2.2 Python版本

4.3 服务器启动脚本

5. 调试技巧与常见问题解决

5.1 调试工具配置

5.1.1 日志系统设置

5.1.2 MCP Inspector使用

5.2 常见问题解决方案

5.3 问题诊断清单

5.4 性能优化建议

6. 进阶功能实现

6.1 资源管理

6.2 提示模板系统

7. 测试与质量保证

7.1 单元测试框架

提示处理器

7.2 集成测试

总结

参考资源


 

摘要

作为一名深耕技术领域多年的博主摘星,我深刻感受到了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, Sequencefrom 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

特点

服务器初始化

new Server()

Server()

类似的构造方式

请求处理

setRequestHandler()

装饰器模式

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, TextContentclass 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 + belif operation == "subtract":result = a - belif operation == "multiply":result = a * belif operation == "divide":if b == 0:raise ValueError("除数不能为零")result = a / belse: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 mainif __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) > 0assert 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应用的发展贡献自己的力量。

参考资源

  • MCP官方文档
  • MCP SDK GitHub仓库
  • Anthropic MCP规范
  • MCP服务器示例集合

🌈 我是摘星!如果这篇文章在你的技术成长路上留下了印记:

👁️ 【关注】与我一起探索技术的无限可能,见证每一次突破

👍 【点赞】为优质技术内容点亮明灯,传递知识的力量

🔖 【收藏】将精华内容珍藏,随时回顾技术要点

💬 【评论】分享你的独特见解,让思维碰撞出智慧火花

🗳️ 【投票】用你的选择为技术社区贡献一份力量

技术路漫漫,让我们携手前行,在代码的世界里摘取属于程序员的那片星辰大海!

http://www.xdnf.cn/news/1218493.html

相关文章:

  • 在幸狐RV1106板子上用gcc14.2本地编译安装samba-4.22.3服务器,并且支持XP系统访问共享文件夹
  • 基于单片机胎压检测/锅炉蒸汽压力/气压检测系统
  • LCM中间件入门(2):LCM核心实现原理解析
  • InfluxDB 与 Python 框架结合:Django 应用案例(二)
  • kmp复习,需要多看多练
  • Kubernetes 应用部署实战:为什么需要 Kubernetes?
  • InfluxDB 与 Python 框架结合:Django 应用案例(三)
  • Java Matcher对象中find()与matches()的区别
  • QT6 Python UI文件转换PY文件的方法
  • HttpServletRequest 和 HttpServletResponse核心接口区别
  • 哈希的概念及其应用
  • linux线程封装和互斥
  • Flutter Chen Generator - yaml配置使用
  • 了解SQL
  • 从姑苏区人工智能大模型基础设施招标|学习服务器、AI处理器、GPU
  • 【车联网kafka】Kafka核心架构与实战经验(第二篇)
  • 防火墙安全实验
  • 《秋招在即!Redis数据类型面试题解析》
  • Vue3+Vite项目如何简单使用tsx
  • SpringBoot+SpringAI打造智能对话机器人
  • MySQL 8.0 OCP 1Z0-908 题目解析(38)
  • Kafka Streams窗口技术全解析:从理论到电商实时分析实战
  • TTS语音合成|GPT-SoVITS语音合成服务器部署,实现http访问
  • Linux多线程线程控制
  • 前端核心技术Node.js(五)——Mongodb、Mongoose和接口
  • 计算机网络学习(一、Cisco Packet Tracer软件安装)
  • 计算机网络学习--------三次握手与四次挥手
  • diffusion原理和代码延伸笔记1——扩散桥,GOUB,UniDB
  • 【计算机网络】5传输层
  • 网络与信息安全有哪些岗位:(4)应急响应工程师