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：基础计算器服务器
from fastmcp import MCPServer, Request
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
mcp_server = MCPServer(app)

class CalculationRequest(BaseModel):
x: float
y: float

@mcp_server.mcp_endpoint
async def add(request: Request[CalculationRequest]) –> float:
"""将两个数字相加"""
data = request.params
return data.x + data.y

@mcp_server.mcp_endpoint
async def subtract(request: Request[CalculationRequest]) –> float:
"""从第一个数字中减去第二个数字"""
data = request.params
return data.x – data.y

@mcp_server.mcp_endpoint
async def multiply(request: Request[CalculationRequest]) –> float:
"""将两个数字相乘"""
data = request.params
return data.x * data.y

@mcp_server.mcp_endpoint
async def divide(request: Request[CalculationRequest]) –> float:
"""将第一个数字除以第二个数字"""
data = request.params
if data.y == 0:
raise ValueError("除数不能为零")
return data.x / data.y

if __name__ == "__main__":
import uvicorn
uvicorn.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, timedelta

app = FastAPI()
mcp_server = MCPServer(app)

class WeatherRequest(BaseModel):
city: str
days: int = 1

class WeatherInfo(BaseModel):
date: str
temperature: float
condition: str
humidity: int

@mcp_server.mcp_endpoint
async 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 result

if __name__ == "__main__":
import uvicorn
uvicorn.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 json

app = FastAPI()
mcp_server = MCPServer(app)

# 模拟文件系统（实际应用中使用真实文件系统）
file_system = {}

class FileWriteRequest(BaseModel):
filename: str
content: str

class FileReadRequest(BaseModel):
filename: str

class FileListRequest(BaseModel):
directory: str = "/"

class FileDeleteRequest(BaseModel):
filename: str

@mcp_server.mcp_endpoint
async def write_file(request: Request[FileWriteRequest]) –> bool:
"""将内容写入文件"""
data = request.params
file_system[data.filename] = data.content
return True

@mcp_server.mcp_endpoint
async 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_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.params
if data.filename not in file_system:
raise ValueError(f"文件 {data.filename} 不存在")
del file_system[data.filename]
return True

if __name__ == "__main__":
import uvicorn
uvicorn.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 random

app = FastAPI()
mcp_server = MCPServer(app)

# 模拟数据存储
notes_db = []
todos_db = []
contacts_db = []

class Note(BaseModel):
id: int = None
title: str
content: str
created_at: str = None

class Todo(BaseModel):
id: int = None
task: str
completed: bool = False
due_date: str = None

class Contact(BaseModel):
id: int = None
name: str
phone: str
email: str = None

class SearchRequest(BaseModel):
query: str

@mcp_server.mcp_endpoint
async 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_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.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_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.params
for todo in todos_db:
if todo.id == todo_id:
todo.completed = True
return todo
raise ValueError(f"待办事项 #{todo_id} 不存在")

@mcp_server.mcp_endpoint
async 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_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 uvicorn
uvicorn.run(app, host="0.0.0.0", port=8003)

小白指南：逐步理解

1. 什么是MCP？

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

2. FastMCP和FastAPI的角色

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

3. 搭建MCP服务器的步骤

实际应用场景

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

最佳实践

结语

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