使用说明
MCP服务器开发框架
概述
一个专为企业级模型上下文协议(MCP)工具开发设计的专业框架,标准化MCP服务器开发过程,帮助开发人员快速构建高质量的AI工具。通过将FastAPI与FastAPI MCP集成,该框架实现了从传统API到AI可调用工具的无缝转换。
主要特点
- MCP工具标准化:自动将传统的FastAPI端点转换为AI模型可调用的MCP工具
- 接口实现分离:支持接口定义和实现之间的明确分离,便于测试和环境切换
- 依赖注入设计:利用FastAPI的依赖注入机制实现灵活的组件组合和解耦
- 完整的开发流程:提供从开发到测试和部署的全面工具链支持
- 示例驱动文档通过实例展示最佳实践,以加速入职流程
该框架特别适用于:
- AI工具开发团队
- 希望将现有API转化为AI工具的开发人员
- 实施标准化微服务架构的组织
建筑
核心架构
┌─────────────┐ ┌───────────────┐ ┌─────────────────┐
│ API Layer │ ──→ │ Service Layer │ ──→ │ Implementation │
└─────────────┘ └───────────────┘ └─────────────────┘
↓
┌─────────────┐
│ MCP Endpoint│ ←── FastAPI-MCP Auto-Conversion
└─────────────┘
关键组件
- FastAPI应用程序:提供HTTP API服务基础
- FastAPI MCP:将API端点转换为MCP工具
- 服务接口层:通过抽象基类定义服务契约
- 依赖注入提供者:管理服务实例创建和注入
- 实现类:包括基于环境的切换的模拟和真实实现
技术栈
- Python 3.10+:利用最新的语言功能和类型注释
- 快速API:高性能异步web框架
- FastAPI MCP:自动将FastAPI端点作为MCP工具公开
- 开发工具链:包括代码质量检查和测试工具
快速开始
环境设置
该框架使用 uv 作为其包管理器,提供更快的依赖关系解析和虚拟环境管理。有关安装说明,请参阅 uv官方文件.
# Install dependencies and set up the development environment
make install
启动示例服务
# Start the development server
make dev
服务运行后,您可以:
- 访问API文档:
http://localhost:5000/docs - 连接到MCP端点
http://localhost:5000/mcp使用MCP客户端(例如Cursor、Claude Desktop)
项目结构
.
├── main.py # Application entry point and route definitions
├── services/ # Service layer implementations
│ └── parking_service.py # Example service (replaceable with custom services)
├── pyproject.toml # Project configuration and dependency definitions
└── Makefile # Development and build tasks
MCP工具开发指南
开发过程概述
- 定义服务接口:创建抽象基类以定义服务契约
- 实施服务:根据需求开发模拟或真实实现
- 创建API端点:使用FastAPI开发API接口
- 启用MCP服务:通过FastAPI MCP将端点作为AI工具公开
服务接口和实现
# 1. Service Interface Definition
from abc import ABC, abstractmethod
from typing import Dict, Any
class DataService(ABC):
@abstractmethod
def get_data(self, id: str) -> Dict[str, Any]:
"""Data retrieval interface"""
pass
# 2A. Mock Implementation - for development and testing
class DataServiceMockImpl(DataService):
def get_data(self, id: str) -> Dict[str, Any]:
return {"id": id, "name": "Test Data", "mock": True}
# 2B. Real Implementation - for production environment
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. Dependency Injection Configuration
def get_data_service() -> DataService:
# Choose implementation based on environment
return DataServiceMockImpl() # or return DataServiceImpl(settings.DATABASE_URL)
API和MCP工具定义
from fastapi import FastAPI, Depends
from fastapi_mcp import FastApiMCP
from pydantic import BaseModel, Field
app = FastAPI()
# Request model
class ItemRequest(BaseModel):
query: str = Field(..., description="Search query parameter")
limit: int = Field(10, description="Result limit")
# API endpoint - automatically converted to an MCP tool
@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"])}
# Create and mount MCP service
mcp = FastApiMCP(
app,
name="example-service",
description="Example MCP service",
base_url="http://localhost:5000",
include_operations=["search_items"]
)
# Mount MCP service at specified path
mcp.mount(mount_path="/mcp")
依赖注入解释
FastAPI的依赖注入系统是该框架不可或缺的一部分,提供了强大而灵活的依赖管理功能。
依赖注入基础
依赖注入是一种设计模式,它允许将依赖关系(如服务、数据库连接等)注入到使用它们的组件中,而不是让组件自己创建和管理依赖关系。在FastAPI中,依赖注入是通过 Depends 功能。
from fastapi import Depends
def get_db():
"""Database connection provider"""
db = connect_to_db()
try:
yield db # Using yield enables lifecycle management of dependencies
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()
- 嵌套依赖项:依赖关系可以依赖于其他依赖关系,形成依赖关系树
MCP工具开发中的应用
在此框架中,依赖注入主要用于:
- 服务实例管理:将服务实现注入API端点
- 环境适应:根据运行时环境选择不同的服务实现
- 资源生命周期管理:管理数据库连接等资源的创建和释放
FastAPI MCP功能
自动生成MCP工具
FastAPI MCP可以自动将FastAPI端点转换为MCP工具:
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI()
# Define a standard FastAPI endpoint
@app.post("/predict", operation_id="predict_sentiment")
async def predict_sentiment(text: str):
return {"sentiment": "positive", "confidence": 0.92}
# Create and mount MCP service - automatically converts the above endpoint to an MCP tool
mcp = FastApiMCP(
app,
name="sentiment-analysis",
description="Sentiment analysis service",
base_url="http://localhost:5000",
include_operations=["predict_sentiment"]
)
# Mount MCP service at specified path
mcp.mount(mount_path="/mcp")
工具命名最佳实践
MCP工具名称默认为API端点的 operation_id。我们建议遵循以下命名约定:
- 使用清晰、描述性的名称
- 采用verb_noun格式(例如。,
predict_sentiment,find_nearby_parking) - 明确设置
operation_id而不是依赖汽车发电
# Recommended: Explicitly set operation_id
@app.post("/parking/nearby", operation_id="find_nearby_parking")
async def find_nearby(request: NearbyRequest):
# Implementation logic...
pass
# Not recommended: Relying on auto-generated operation_id (generates something like "find_nearby_parking_nearby_post")
@app.post("/parking/nearby")
async def find_nearby(request: NearbyRequest):
# Implementation logic...
pass
测试和质量保证
该框架支持多级测试策略:
# Run code quality checks
make check
# Run test suite
make test
测试策略
- 单元测试:使用Mock服务实现测试单个组件
- 集成测试:测试组件之间的相互作用
- API测试:验证HTTP接口行为
- MCP工具测试:验证AI工具功能
性能优化
为确保MCP工具的高性能,我们建议:
- 异步处理:利用FastAPI的异步功能进行并发请求处理
- 缓存策略:对频繁请求的数据实施缓存
- 批处理:设计支持批处理操作的API,以减少调用频率
贡献指南
欢迎对这一框架作出贡献:
- 复刻仓库
- 创建要素分支(
git checkout -b feature/amazing-feature) - 提交您的更改(
git commit -m 'Add amazing feature') - 推到分支(
git push origin feature/amazing-feature) - 创建拉取请求
许可证
参考文献
- 快速API -高性能web框架
- FastAPI MCP -MCP工具自动生成库