网站导航
> 技术文档 > FastMCP与FastAPI：构建自定义MCP服务器

FastMCP与FastAPI：构建自定义MCP服务器

网友: 技术文档 2025-08-15 16:07:02

FastMCP与FastAPI：构建自定义MCP服务器

模型上下文协议(Model Context Protocol, MCP)是一种让AI模型与外部工具和服务交互的标准。本文将介绍FastMCP和FastAPI，并通过实例展示如何创建自定义MCP服务器。

MCP基础概念

MCP允许语言模型:

  • 访问外部工具和API
  • 执行实时计算和查询
  • 与文件系统和服务交互

简单来说，MCP让AI模型能\"走出\"对话框，调用各种功能。

FastMCP介绍

FastMCP是一个Python库，简化了MCP服务器的构建。它提供了:

  • 易用的装饰器语法
  • 自动处理请求/响应
  • 参数验证和错误处理

FastAPI介绍

FastAPI是一个现代Web框架，用于构建API：

  • 高性能（基于ASGI）
  • 自动生成交互式文档
  • 数据验证和序列化
  • 基于Python类型提示

FastMCP与FastAPI的结合

二者结合的优势：

  • FastAPI提供了Web服务器基础架构
  • FastMCP添加了模型交互能力
  • 共享相似的装饰器语法和类型系统

自定义MCP服务器示例

# 示例1：基础计算器服务器from fastmcp import MCPServer, Requestfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()mcp_server = MCPServer(app)class CalculationRequest(BaseModel): x: float y: float@mcp_server.mcp_endpointasync def add(request: Request[CalculationRequest]) -> float: \"\"\"将两个数字相加\"\"\" data = request.params return data.x + data.y@mcp_server.mcp_endpointasync def subtract(request: Request[CalculationRequest]) -> float: \"\"\"从第一个数字中减去第二个数字\"\"\" data = request.params return data.x - data.y@mcp_server.mcp_endpointasync def multiply(request: Request[CalculationRequest]) -> float: \"\"\"将两个数字相乘\"\"\" data = request.params return data.x * data.y@mcp_server.mcp_endpointasync def divide(request: Request[CalculationRequest]) -> float: \"\"\"将第一个数字除以第二个数字\"\"\" data = request.params if data.y == 0: raise ValueError(\"除数不能为零\") return data.x / data.yif __name__ == \"__main__\": import uvicorn uvicorn.run(app, host=\"0.0.0.0\", port=8000)# 示例2：天气信息服务器from fastmcp import MCPServer, Requestfrom fastapi import FastAPIfrom pydantic import BaseModelimport randomfrom datetime import datetime, timedeltaapp = FastAPI()mcp_server = MCPServer(app)class WeatherRequest(BaseModel): city: str days: int = 1class WeatherInfo(BaseModel): date: str temperature: float condition: str humidity: int@mcp_server.mcp_endpointasync def get_weather(request: Request[WeatherRequest]) -> list[WeatherInfo]: \"\"\"获取指定城市的天气预报\"\"\" data = request.params # 这里使用模拟数据，实际应用中会调用真实的天气API weather_conditions = [\"晴朗\", \"多云\", \"小雨\", \"大雨\", \"雷雨\", \"小雪\"] result = [] today = datetime.now() for i in range(data.days): date = today + timedelta(days=i) result.append(WeatherInfo( date=date.strftime(\"%Y-%m-%d\"), temperature=round(random.uniform(15, 30), 1), condition=random.choice(weather_conditions), humidity=random.randint(30, 90) )) return resultif __name__ == \"__main__\": import uvicorn uvicorn.run(app, host=\"0.0.0.0\", port=8001)# 示例3：文件操作服务器from fastmcp import MCPServer, Requestfrom fastapi import FastAPIfrom pydantic import BaseModelimport osimport jsonapp = FastAPI()mcp_server = MCPServer(app)# 模拟文件系统（实际应用中使用真实文件系统）file_system = {}class FileWriteRequest(BaseModel): filename: str content: strclass FileReadRequest(BaseModel): filename: strclass FileListRequest(BaseModel): directory: str = \"/\"class FileDeleteRequest(BaseModel): filename: str@mcp_server.mcp_endpointasync def write_file(request: Request[FileWriteRequest]) -> bool: \"\"\"将内容写入文件\"\"\" data = request.params file_system[data.filename] = data.content return True@mcp_server.mcp_endpointasync def read_file(request: Request[FileReadRequest]) -> str: \"\"\"读取文件内容\"\"\" data = request.params if data.filename not in file_system: raise ValueError(f\"文件 {data.filename} 不存在\") return file_system[data.filename]@mcp_server.mcp_endpointasync def list_files(request: Request[FileListRequest]) -> list[str]: \"\"\"列出指定目录中的文件\"\"\" return list(file_system.keys())@mcp_server.mcp_endpointasync def delete_file(request: Request[FileDeleteRequest]) -> bool: \"\"\"删除文件\"\"\" data = request.params if data.filename not in file_system: raise ValueError(f\"文件 {data.filename} 不存在\") del file_system[data.filename] return Trueif __name__ == \"__main__\": import uvicorn uvicorn.run(app, host=\"0.0.0.0\", port=8002)# 示例4：综合应用 - 个人助手服务from fastmcp import MCPServer, Requestfrom fastapi import FastAPIfrom pydantic import BaseModelfrom datetime import datetimeimport jsonimport randomapp = FastAPI()mcp_server = MCPServer(app)# 模拟数据存储notes_db = []todos_db = []contacts_db = []class Note(BaseModel): id: int = None title: str content: str created_at: str = Noneclass Todo(BaseModel): id: int = None task: str completed: bool = False due_date: str = Noneclass Contact(BaseModel): id: int = None name: str phone: str email: str = Noneclass SearchRequest(BaseModel): query: str@mcp_server.mcp_endpointasync def add_note(request: Request[Note]) -> Note: \"\"\"添加一条笔记\"\"\" note = request.params note.id = len(notes_db) + 1 note.created_at = datetime.now().isoformat() notes_db.append(note) return note@mcp_server.mcp_endpointasync def get_notes(request: Request) -> list[Note]: \"\"\"获取所有笔记\"\"\" return notes_db@mcp_server.mcp_endpointasync def add_todo(request: Request[Todo]) -> Todo: \"\"\"添加一个待办事项\"\"\" todo = request.params todo.id = len(todos_db) + 1 if not todo.due_date: todo.due_date = (datetime.now() + timedelta(days=1)).isoformat() todos_db.append(todo) return todo@mcp_server.mcp_endpointasync def get_todos(request: Request) -> list[Todo]: \"\"\"获取所有待办事项\"\"\" return todos_db@mcp_server.mcp_endpointasync def complete_todo(request: Request[int]) -> Todo: \"\"\"标记待办事项为已完成\"\"\" todo_id = request.params for todo in todos_db: if todo.id == todo_id: todo.completed = True return todo raise ValueError(f\"待办事项 #{todo_id} 不存在\")@mcp_server.mcp_endpointasync def add_contact(request: Request[Contact]) -> Contact: \"\"\"添加联系人\"\"\" contact = request.params contact.id = len(contacts_db) + 1 contacts_db.append(contact) return contact@mcp_server.mcp_endpointasync def get_contacts(request: Request) -> list[Contact]: \"\"\"获取所有联系人\"\"\" return contacts_db@mcp_server.mcp_endpointasync def search(request: Request[SearchRequest]) -> dict: \"\"\"搜索笔记、待办事项和联系人\"\"\" query = request.params.query.lower() matching_notes = [note for note in notes_db if query in note.title.lower() or query in note.content.lower()] matching_todos = [todo for todo in todos_db if query in todo.task.lower()] matching_contacts = [contact for contact in contacts_db if query in contact.name.lower()] return { \"notes\": matching_notes, \"todos\": matching_todos, \"contacts\": matching_contacts }if __name__ == \"__main__\": import uvicorn uvicorn.run(app, host=\"0.0.0.0\", port=8003)

小白指南：逐步理解

1. 什么是MCP？

想象你的AI助手（如ChatGPT）是一个聪明的专家，但被关在一个房间里，只能通过纸条与外界交流。MCP就像是给这个专家配备了一部电话，让它能打电话给各种服务：查天气、计算数学问题、管理你的日程等。

2. FastMCP和FastAPI的角色

  • FastAPI：提供电话线路和基础设施
  • FastMCP：定义通话协议，确保AI能正确拨号和理解回复

3. 搭建MCP服务器的步骤

  1. 安装必要的库：

    pip install fastmcp fastapi uvicorn

  2. 创建服务器框架：

    from fastmcp import MCPServerfrom fastapi import FastAPIapp = FastAPI()mcp_server = MCPServer(app)

  3. 定义数据模型（请求和响应的结构）

  4. 创建功能端点（用@mcp_server.mcp_endpoint装饰器）

  5. 启动服务器：

    import uvicornuvicorn.run(app, host=\"0.0.0.0\", port=8000)

实际应用场景

  • 数据分析助手：连接数据处理工具
  • 客户服务机器人：访问CRM系统和知识库
  • 智能文档助手：处理和生成各类文档
  • 个人生产力工具：管理日程、笔记和待办事项

最佳实践

  1. 明确定义数据模型：使用Pydantic确保输入输出格式正确
  2. 添加详细文档：每个端点都应有清晰的描述
  3. 实现错误处理：优雅处理异常情况
  4. 使用异步功能：充分利用FastAPI的异步特性
  5. 添加验证和安全措施：保护服务器不受恶意请求攻击

结语

FastMCP和FastAPI的结合为AI模型提供了强大的扩展能力。通过本文介绍的示例，即使是编程新手也能构建自己的MCP服务器，让AI助手拥有更多实用功能。随着MCP生态系统的发展，我们可以期待更多创新应用的出现。

网络标签：

相关问题