MCP服务器开发框架

一个为企业级模型上下文协议(MCP)工具开发设计的专业框架,通过集成FastAPI与FastAPI-MCP,实现传统API到AI可调用工具的无缝转换。

  • by mlzoo
  • Python
  • 本地部署
  • 208 星标
  • 更新于
在 GitHub 查看 mlzoo/mcp_forge

使用说明

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工具开发指南

开发过程概述

  1. 定义服务接口:创建抽象基类以定义服务契约
  2. 实施服务:根据需求开发模拟或真实实现
  3. 创建API端点:使用FastAPI开发API接口
  4. 启用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支持多种类型的依赖注入:

  1. 函数依赖:如上所示,使用函数作为依赖提供者
  2. 类依赖关系:将类用作更复杂的依赖关系管理的依赖关系
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()
  1. 嵌套依赖项:依赖关系可以依赖于其他依赖关系,形成依赖关系树

MCP工具开发中的应用

在此框架中,依赖注入主要用于:

  1. 服务实例管理:将服务实现注入API端点
  2. 环境适应:根据运行时环境选择不同的服务实现
  3. 资源生命周期管理:管理数据库连接等资源的创建和释放

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,以减少调用频率

贡献指南

欢迎对这一框架作出贡献:

  1. 复刻仓库
  2. 创建要素分支(git checkout -b feature/amazing-feature)
  3. 提交您的更改(git commit -m 'Add amazing feature')
  4. 推到分支(git push origin feature/amazing-feature)
  5. 创建拉取请求

许可证

MIT许可证

参考文献