FastMCP与FastAPI：构建自界说MCP服务器

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. # 示例1：基础计算器服务器
2. from fastmcp import MCPServer, Request
3. from fastapi import FastAPI
4. from pydantic import BaseModel
6. app = FastAPI()
7. mcp_server = MCPServer(app)
9. class CalculationRequest(BaseModel):
10.     x: float
11.     y: float
13. @mcp_server.mcp_endpoint
14. async def add(request: Request[CalculationRequest]) -> float:
15.     """将两个数字相加"""
16.     data = request.params
17.     return data.x + data.y
19. @mcp_server.mcp_endpoint
20. async def subtract(request: Request[CalculationRequest]) -> float:
21.     """从第一个数字中减去第二个数字"""
22.     data = request.params
23.     return data.x - data.y
25. @mcp_server.mcp_endpoint
26. async def multiply(request: Request[CalculationRequest]) -> float:
27.     """将两个数字相乘"""
28.     data = request.params
29.     return data.x * data.y
31. @mcp_server.mcp_endpoint
32. async def divide(request: Request[CalculationRequest]) -> float:
33.     """将第一个数字除以第二个数字"""
34.     data = request.params
35.     if data.y == 0:
36.         raise ValueError("除数不能为零")
37.     return data.x / data.y
39. if __name__ == "__main__":
40.     import uvicorn
41.     uvicorn.run(app, host="0.0.0.0", port=8000)
44. # 示例2：天气信息服务器
45. from fastmcp import MCPServer, Request
46. from fastapi import FastAPI
47. from pydantic import BaseModel
48. import random
49. from datetime import datetime, timedelta
51. app = FastAPI()
52. mcp_server = MCPServer(app)
54. class WeatherRequest(BaseModel):
55.     city: str
56.     days: int = 1
58. class WeatherInfo(BaseModel):
59.     date: str
60.     temperature: float
61.     condition: str
62.     humidity: int
64. @mcp_server.mcp_endpoint
65. async def get_weather(request: Request[WeatherRequest]) -> list[WeatherInfo]:
66.     """获取指定城市的天气预报"""
67.     data = request.params
68.    
69.     # 这里使用模拟数据，实际应用中会调用真实的天气API
70.     weather_conditions = ["晴朗", "多云", "小雨", "大雨", "雷雨", "小雪"]
71.    
72.     result = []
73.     today = datetime.now()
74.    
75.     for i in range(data.days):
76.         date = today + timedelta(days=i)
77.         result.append(WeatherInfo(
78.             date=date.strftime("%Y-%m-%d"),
79.             temperature=round(random.uniform(15, 30), 1),
80.             condition=random.choice(weather_conditions),
81.             humidity=random.randint(30, 90)
82.         ))
83.    
84.     return result
86. if __name__ == "__main__":
87.     import uvicorn
88.     uvicorn.run(app, host="0.0.0.0", port=8001)
91. # 示例3：文件操作服务器
92. from fastmcp import MCPServer, Request
93. from fastapi import FastAPI
94. from pydantic import BaseModel
95. import os
96. import json
98. app = FastAPI()
99. mcp_server = MCPServer(app)
101. # 模拟文件系统（实际应用中使用真实文件系统）
102. file_system = {}
104. class FileWriteRequest(BaseModel):
105.     filename: str
106.     content: str
108. class FileReadRequest(BaseModel):
109.     filename: str
111. class FileListRequest(BaseModel):
112.     directory: str = "/"
114. class FileDeleteRequest(BaseModel):
115.     filename: str
117. @mcp_server.mcp_endpoint
118. async def write_file(request: Request[FileWriteRequest]) -> bool:
119.     """将内容写入文件"""
120.     data = request.params
121.     file_system[data.filename] = data.content
122.     return True
124. @mcp_server.mcp_endpoint
125. async def read_file(request: Request[FileReadRequest]) -> str:
126.     """读取文件内容"""
127.     data = request.params
128.     if data.filename not in file_system:
129.         raise ValueError(f"文件 {data.filename} 不存在")
130.     return file_system[data.filename]
132. @mcp_server.mcp_endpoint
133. async def list_files(request: Request[FileListRequest]) -> list[str]:
134.     """列出指定目录中的文件"""
135.     return list(file_system.keys())
137. @mcp_server.mcp_endpoint
138. async def delete_file(request: Request[FileDeleteRequest]) -> bool:
139.     """删除文件"""
140.     data = request.params
141.     if data.filename not in file_system:
142.         raise ValueError(f"文件 {data.filename} 不存在")
143.     del file_system[data.filename]
144.     return True
146. if __name__ == "__main__":
147.     import uvicorn
148.     uvicorn.run(app, host="0.0.0.0", port=8002)
151. # 示例4：综合应用 - 个人助手服务
152. from fastmcp import MCPServer, Request
153. from fastapi import FastAPI
154. from pydantic import BaseModel
155. from datetime import datetime
156. import json
157. import random
159. app = FastAPI()
160. mcp_server = MCPServer(app)
162. # 模拟数据存储
163. notes_db = []
164. todos_db = []
165. contacts_db = []
167. class Note(BaseModel):
168.     id: int = None
169.     title: str
170.     content: str
171.     created_at: str = None
173. class Todo(BaseModel):
174.     id: int = None
175.     task: str
176.     completed: bool = False
177.     due_date: str = None
179. class Contact(BaseModel):
180.     id: int = None
181.     name: str
182.     phone: str
183.     email: str = None
185. class SearchRequest(BaseModel):
186.     query: str
188. @mcp_server.mcp_endpoint
189. async def add_note(request: Request[Note]) -> Note:
190.     """添加一条笔记"""
191.     note = request.params
192.     note.id = len(notes_db) + 1
193.     note.created_at = datetime.now().isoformat()
194.     notes_db.append(note)
195.     return note
197. @mcp_server.mcp_endpoint
198. async def get_notes(request: Request) -> list[Note]:
199.     """获取所有笔记"""
200.     return notes_db
202. @mcp_server.mcp_endpoint
203. async def add_todo(request: Request[Todo]) -> Todo:
204.     """添加一个待办事项"""
205.     todo = request.params
206.     todo.id = len(todos_db) + 1
207.     if not todo.due_date:
208.         todo.due_date = (datetime.now() + timedelta(days=1)).isoformat()
209.     todos_db.append(todo)
210.     return todo
212. @mcp_server.mcp_endpoint
213. async def get_todos(request: Request) -> list[Todo]:
214.     """获取所有待办事项"""
215.     return todos_db
217. @mcp_server.mcp_endpoint
218. async def complete_todo(request: Request[int]) -> Todo:
219.     """标记待办事项为已完成"""
220.     todo_id = request.params
221.     for todo in todos_db:
222.         if todo.id == todo_id:
223.             todo.completed = True
224.             return todo
225.     raise ValueError(f"待办事项 #{todo_id} 不存在")
227. @mcp_server.mcp_endpoint
228. async def add_contact(request: Request[Contact]) -> Contact:
229.     """添加联系人"""
230.     contact = request.params
231.     contact.id = len(contacts_db) + 1
232.     contacts_db.append(contact)
233.     return contact
235. @mcp_server.mcp_endpoint
236. async def get_contacts(request: Request) -> list[Contact]:
237.     """获取所有联系人"""
238.     return contacts_db
240. @mcp_server.mcp_endpoint
241. async def search(request: Request[SearchRequest]) -> dict:
242.     """搜索笔记、待办事项和联系人"""
243.     query = request.params.query.lower()
244.    
245.     matching_notes = [note for note in notes_db if query in note.title.lower() or query in note.content.lower()]
246.     matching_todos = [todo for todo in todos_db if query in todo.task.lower()]
247.     matching_contacts = [contact for contact in contacts_db if query in contact.name.lower()]
248.    
249.     return {
250.         "notes": matching_notes,
251.         "todos": matching_todos,
252.         "contacts": matching_contacts
253.     }
255. if __name__ == "__main__":
256.     import uvicorn
257.     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

  复制代码
• 创建服务器框架：
  1. from fastmcp import MCPServer
  2. from fastapi import FastAPI
  4. app = FastAPI()
  5. mcp_server = MCPServer(app)

  复制代码
• 界说数据模子（哀求和相应的结构）
  
• 建立功能端点（用@mcp_server.mcp_endpoint装饰器）
  
• 启动服务器：
  1. import uvicorn
  2. uvicorn.run(app, host="0.0.0.0", port=8000)

  复制代码

现实应用场景

• 数据分析助手：毗连数据处置惩罚工具
  
• 客户服务呆板人：访问CRM体系和知识库
  
• 智能文档助手：处置惩罚和天生各类文档
  
• 个人生产力工具：管理日程、条记和待服务项
  

最佳实践

• 明白界说数据模子：利用Pydantic确保输入输出格式准确
  
• 添加具体文档：每个端点都应有清楚的形貌
  
• 实现错误处置惩罚：优雅处置惩罚异常环境
  
• 利用异步功能：充分利用FastAPI的异步特性
  
• 添加验证和安全步调：掩护服务器不受恶意哀求攻击
  

结语

FastMCP和FastAPI的联合为AI模子提供了强大的扩展能力。通过本文先容的示例，纵然是编程新手也能构建自己的MCP服务器，让AI助手拥有更多实勤劳能。随着MCP生态体系的发展，我们可以等候更多创新应用的出现。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。