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

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

news 来源：原创 2025/4/25 5:26:32

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, Request
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()
mcp_server = MCPServer(app)class CalculationRequest(BaseModel):x: floaty: float@mcp_server.mcp_endpoint
async def add(request: Request[CalculationRequest]) -> float:"""将两个数字相加"""data = request.paramsreturn data.x + data.y@mcp_server.mcp_endpoint
async def subtract(request: Request[CalculationRequest]) -> float:"""从第一个数字中减去第二个数字"""data = request.paramsreturn data.x - data.y@mcp_server.mcp_endpoint
async def multiply(request: Request[CalculationRequest]) -> float:"""将两个数字相乘"""data = request.paramsreturn data.x * data.y@mcp_server.mcp_endpoint
async def divide(request: Request[CalculationRequest]) -> float:"""将第一个数字除以第二个数字"""data = request.paramsif data.y == 0:raise ValueError("除数不能为零")return data.x / data.yif __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)# 示例2：天气信息服务器
from fastmcp import MCPServer, Request
from fastapi import FastAPI
from pydantic import BaseModel
import random
from datetime import datetime, timedeltaapp = FastAPI()
mcp_server = MCPServer(app)class WeatherRequest(BaseModel):city: strdays: int = 1class WeatherInfo(BaseModel):date: strtemperature: floatcondition: strhumidity: int@mcp_server.mcp_endpoint
async def get_weather(request: Request[WeatherRequest]) -> list[WeatherInfo]:"""获取指定城市的天气预报"""data = request.params# 这里使用模拟数据，实际应用中会调用真实的天气APIweather_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 uvicornuvicorn.run(app, host="0.0.0.0", port=8001)# 示例3：文件操作服务器
from fastmcp import MCPServer, Request
from fastapi import FastAPI
from pydantic import BaseModel
import os
import jsonapp = FastAPI()
mcp_server = MCPServer(app)# 模拟文件系统（实际应用中使用真实文件系统）
file_system = {}class FileWriteRequest(BaseModel):filename: strcontent: strclass FileReadRequest(BaseModel):filename: strclass FileListRequest(BaseModel):directory: str = "/"class FileDeleteRequest(BaseModel):filename: str@mcp_server.mcp_endpoint
async def write_file(request: Request[FileWriteRequest]) -> bool:"""将内容写入文件"""data = request.paramsfile_system[data.filename] = data.contentreturn True@mcp_server.mcp_endpoint
async def read_file(request: Request[FileReadRequest]) -> str:"""读取文件内容"""data = request.paramsif data.filename not in file_system:raise ValueError(f"文件 {data.filename} 不存在")return file_system[data.filename]@mcp_server.mcp_endpoint
async def list_files(request: Request[FileListRequest]) -> list[str]:"""列出指定目录中的文件"""return list(file_system.keys())@mcp_server.mcp_endpoint
async def delete_file(request: Request[FileDeleteRequest]) -> bool:"""删除文件"""data = request.paramsif data.filename not in file_system:raise ValueError(f"文件 {data.filename} 不存在")del file_system[data.filename]return Trueif __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8002)# 示例4：综合应用 - 个人助手服务
from fastmcp import MCPServer, Request
from fastapi import FastAPI
from pydantic import BaseModel
from datetime import datetime
import json
import randomapp = FastAPI()
mcp_server = MCPServer(app)# 模拟数据存储
notes_db = []
todos_db = []
contacts_db = []class Note(BaseModel):id: int = Nonetitle: strcontent: strcreated_at: str = Noneclass Todo(BaseModel):id: int = Nonetask: strcompleted: bool = Falsedue_date: str = Noneclass Contact(BaseModel):id: int = Nonename: strphone: stremail: str = Noneclass SearchRequest(BaseModel):query: str@mcp_server.mcp_endpoint
async def add_note(request: Request[Note]) -> Note:"""添加一条笔记"""note = request.paramsnote.id = len(notes_db) + 1note.created_at = datetime.now().isoformat()notes_db.append(note)return note@mcp_server.mcp_endpoint
async def get_notes(request: Request) -> list[Note]:"""获取所有笔记"""return notes_db@mcp_server.mcp_endpoint
async def add_todo(request: Request[Todo]) -> Todo:"""添加一个待办事项"""todo = request.paramstodo.id = len(todos_db) + 1if not todo.due_date:todo.due_date = (datetime.now() + timedelta(days=1)).isoformat()todos_db.append(todo)return todo@mcp_server.mcp_endpoint
async def get_todos(request: Request) -> list[Todo]:"""获取所有待办事项"""return todos_db@mcp_server.mcp_endpoint
async def complete_todo(request: Request[int]) -> Todo:"""标记待办事项为已完成"""todo_id = request.paramsfor todo in todos_db:if todo.id == todo_id:todo.completed = Truereturn todoraise ValueError(f"待办事项 #{todo_id} 不存在")@mcp_server.mcp_endpoint
async def add_contact(request: Request[Contact]) -> Contact:"""添加联系人"""contact = request.paramscontact.id = len(contacts_db) + 1contacts_db.append(contact)return contact@mcp_server.mcp_endpoint
async def get_contacts(request: Request) -> list[Contact]:"""获取所有联系人"""return contacts_db@mcp_server.mcp_endpoint
async 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 uvicornuvicorn.run(app, host="0.0.0.0", port=8003)

小白指南：逐步理解

1. 什么是MCP？

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

2. FastMCP和FastAPI的角色

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

3. 搭建MCP服务器的步骤

安装必要的库：
```
pip install fastmcp fastapi uvicorn
```

创建服务器框架：

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

定义数据模型（请求和响应的结构）
创建功能端点（用@mcp_server.mcp_endpoint装饰器）

启动服务器：

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

实际应用场景

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

最佳实践

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

结语

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

入门 Go 语言

Java 2025 技术全景与实战指南：从新特性到架构革新

十分钟恢复服务器攻击——群联AI云防护系统实战

LLMind：利用大型语言模型协调人工智能与物联网以执行复杂任务

全行业软件定制：APP/小程序/系统开发与物联网解决方案

面阵相机中M12镜头和远心镜头的区别及性能优势

Linux下编译opencv-4.10.0（静态链接库和动态链接库）

Vivado版本升级后AXI4-Stream Data FIFO端口变化

centos7使用yum快速安装最新版本Jenkins-2.462.3

IDEA将本地的JAR文件手动安装到 Maven的本地仓库

AI日报 - 2025年04月25日

数据库day-07

VS Code 智能代理模式：重塑开发体验

NOIP2013 提高组.转圈游戏

Explain详解与索引最佳实践

利用deepseek快速生成甘特图

32单片机——外部中断

基于 springboot+vue+elementui 的办公自动化系统设计(

代码注释标记的含义

图论算法体系：并查集、生成树、排序与路径搜索全解析

韩国检方重启调查金建希操纵股价案

财政部部长：中方主张通过平等对话协商解决贸易和关税争议

国防部就美军“压力测试”大演习答澎湃：中国从来不信邪，不怕打，不怕压

湖南永州公安全面推行“项目警官制”，为重点项目建设护航

2025年中央金融机构注资特别国债发行，发行金额1650亿

聚焦“共赢蓝色未来” “海洋命运共同体”上海论坛举行

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

MCP基础概念

FastMCP介绍

FastAPI介绍

FastMCP与FastAPI的结合

自定义MCP服务器示例

小白指南：逐步理解

1. 什么是MCP？

2. FastMCP和FastAPI的角色

3. 搭建MCP服务器的步骤

实际应用场景

最佳实践

结语

相关文章：