一个专为企业级模型上下文协议(MCP)工具开发设计的专业框架,标准化了MCP服务器的开发流程,帮助开发者快速构建高质量的人工智能工具。通过集成FastAPI与FastAPI-MCP,该框架实现了从传统API到AI可调用工具的无缝转换。
此框架特别适用于:
┌─────────────┐ ┌───────────────┐ ┌─────────────────┐
│ API 层 │ ──→ │ 服务层 │ ──→ │ 实现层 │
└─────────────┘ └───────────────┘ └─────────────────┘
↓
┌─────────────┐
│ MCP 端点 │ ←── FastAPI-MCP 自动转换
└─────────────┘
此框架使用uv作为其包管理器,提供更快的依赖解析和虚拟环境管理。安装说明请参阅uv官方文档。
# 安装依赖并设置开发环境
make install
# 启动开发服务器
make dev
服务运行后,您可以:
http://localhost:5000/docshttp://localhost:5000/mcp.
├── main.py # 应用入口点和路由定义
├── services/ # 服务层实现
│ └── parking_service.py # 示例服务(可替换为自定义服务)
├── pyproject.toml # 项目配置和依赖定义
└── Makefile # 开发和构建任务
# 1. 服务接口定义
from abc import ABC, abstractmethod
from typing import Dict, Any
class DataService(ABC):
@abstractmethod
def get_data(self, id: str) -> Dict[str, Any]:
"""数据检索接口"""
pass
# 2A. 模拟实现 - 用于开发和测试
class DataServiceMockImpl(DataService):
def get_data(self, id: str) -> Dict[str, Any]:
return {"id": id, "name": "测试数据", "mock": True}
# 2B. 真实实现 - 用于生产环境
class DataServiceImpl(DataService):
def __init__(self, database_url: str):
self.db = Database(database_url)
def get_data(self, id: str) -> Dict[str, Any]:
return self.db.query("SELECT * FROM data WHERE id = :id", {"id": id})
# 3. 依赖注入配置
def get_data_service() -> DataService:
# 根据环境选择实现
return DataServiceMockImpl() # 或返回 DataServiceImpl(settings.DATABASE_URL)
from fastapi import FastAPI, Depends
from fastapi_mcp import FastApiMCP
from pydantic import BaseModel, Field
app = FastAPI()
# 请求模型
class ItemRequest(BaseModel):
query: str = Field(..., description="搜索查询参数")
limit: int = Field(10, description="结果限制")
# API端点 - 自动转换为MCP工具
@app.post("/items/search", operation_id="search_items")
async def search_items(
request: ItemRequest,
service: DataService = Depends(get_data_service)
):
result = service.search_items(request.query, request.limit)
return {"items": result["items"], "total": len(result["items"])}
# 创建并挂载MCP服务
mcp = FastApiMCP(
app,
name="example-service",
description="示例MCP服务",
base_url="http://localhost:5000",
include_operations=["search_items"]
)
# 在指定路径挂载MCP服务
mcp.mount(mount_path="/mcp")
FastAPI的依赖注入系统是此框架的重要组成部分,提供了强大且灵活的依赖管理能力。
依赖注入是一种设计模式,允许依赖项(如服务、数据库连接等)被注入到使用它们的组件中,而不是由组件自己创建和管理依赖项。在FastAPI中,依赖注入通过Depends函数实现。
from fastapi import Depends
def get_db():
"""数据库连接提供者"""
db = connect_to_db()
try:
yield db # 使用yield可以管理依赖项的生命周期
finally:
db.close()
@app.get("/items/")
async def get_items(db = Depends(get_db)):
return db.query(Item).all()
FastAPI支持多种类型的依赖注入:
class DatabaseDependency:
def __init__(self, settings = Depends(get_settings)):
self.settings = settings
def __call__(self):
db = connect_to_db(self.settings.db_url)
try:
yield db
finally:
db.close()
@app.get("/users/")
async def get_users(db = Depends(DatabaseDependency())):
return db.query(User).all()
在此框架中,依赖注入主要用于:
FastAPI-MCP可以自动将FastAPI端点转换为MCP工具:
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI()
# 定义标准的FastAPI端点
@app.post("/predict", operation_id="predict_sentiment")
async def predict_sentiment(text: str):
return {"sentiment": "positive", "confidence": 0.92}
# 创建并挂载MCP服务 - 自动将上述端点转换为MCP工具
mcp = FastApiMCP(
app,
name="sentiment-analysis",
description="情感分析服务",
base_url="http://localhost:5000",
include_operations=["predict_sentiment"]
)
# 在指定路径挂载MCP服务
mcp.mount(mount_path="/mcp")
MCP工具名称默认为API端点的operation_id。我们建议遵循以下命名约定:
predict_sentiment,find_nearby_parking)operation_id,而不是依赖于自动生成# 推荐:显式设置operation_id
@app.post("/parking/nearby", operation_id="find_nearby_parking")
async def find_nearby(request: NearbyRequest):
# 实现逻辑...
pass
# 不推荐:依赖自动生成的operation_id(生成类似"find_nearby_parking_nearby_post"的东西)
@app.post("/parking/nearby")
async def find_nearby(request: NearbyRequest):
# 实现逻辑...
pass
此框架支持多层级测试策略:
# 运行代码质量检查
make check
# 运行测试套件
make test
为了确保MCP工具的高性能,我们建议:
欢迎对此框架进行贡献:
git checkout -b feature/amazing-feature)git commit -m '添加惊人的功能')git push origin feature/amazing-feature)