返回市场
MCP锻造服务器

MCP锻造服务器

作者:mlzoo178 星标更新:2025-04-26

项目介绍

MCP服务器开发框架

中文文档

概述

一个专为企业级模型上下文协议(MCP)工具开发设计的专业框架,标准化了MCP服务器的开发流程,帮助开发者快速构建高质量的人工智能工具。通过集成FastAPI与FastAPI-MCP,该框架实现了从传统API到AI可调用工具的无缝转换。

关键特性

  • MCP工具标准化:自动将传统的FastAPI端点转换为AI模型可调用的MCP工具
  • 接口实现分离:支持清晰的接口定义与实现分离,便于测试和环境切换
  • 依赖注入设计:利用FastAPI的依赖注入机制,实现灵活的组件组合与解耦
  • 完整的开发流水线:提供从开发到测试和部署的全面工具链支持
  • 示例驱动文档:通过实际示例展示最佳实践,加速上手过程

此框架特别适用于:

  • AI工具开发团队
  • 希望将现有API转换为AI工具的开发者
  • 实施标准化微服务架构的组织

架构

核心架构

┌─────────────┐      ┌───────────────┐      ┌─────────────────┐
│   API 层   │ ──→  │ 服务层         │ ──→  │ 实现层           │
└─────────────┘      └───────────────┘      └─────────────────┘
       ↓
┌─────────────┐
│ MCP 端点   │ ←── FastAPI-MCP 自动转换
└─────────────┘

关键组件

  • FastAPI 应用程序:提供HTTP API服务基础
  • FastAPI-MCP:将API端点转换为MCP工具
  • 服务接口层:通过抽象基类定义服务契约
  • 依赖注入提供者:管理服务实例的创建和注入
  • 实现类:包括基于环境切换的模拟和真实实现

技术栈

  • Python 3.10+:利用最新语言特性和类型注解
  • FastAPI:高性能异步Web框架
  • FastAPI-MCP:自动暴露FastAPI端点为MCP工具
  • 开发工具链:包括代码质量检查和测试工具

快速开始

环境设置

此框架使用uv作为其包管理器,提供更快的依赖解析和虚拟环境管理。安装说明请参阅uv官方文档

# 安装依赖并设置开发环境
make install

启动示例服务

# 启动开发服务器
make dev

服务运行后,您可以:

  • 访问API文档 http://localhost:5000/docs
  • 使用MCP客户端(如Cursor,Claude Desktop)连接到MCP端点 http://localhost:5000/mcp

项目结构

.
├── main.py                    # 应用入口点和路由定义
├── services/                  # 服务层实现
│   └── parking_service.py     # 示例服务(可替换为自定义服务)
├── pyproject.toml             # 项目配置和依赖定义
└── Makefile                   # 开发和构建任务

MCP工具开发指南

开发流程概述

  1. 定义服务接口:创建抽象基类以定义服务契约
  2. 实现服务:根据需求开发模拟或真实实现
  3. 创建API端点:使用FastAPI开发API接口
  4. 启用MCP服务:通过FastAPI-MCP将端点暴露为AI工具

服务接口和实现

# 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)

API和MCP工具定义

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支持多种类型的依赖注入:

  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()

# 定义标准的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_sentimentfind_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

测试策略

  • 单元测试:使用模拟服务实现测试单个组件
  • 集成测试:测试组件之间的交互
  • API测试:验证HTTP接口行为
  • MCP工具测试:验证AI工具功能

性能优化

为了确保MCP工具的高性能,我们建议:

  • 异步处理:利用FastAPI的异步特性进行并发请求处理
  • 缓存策略:为频繁请求的数据实施缓存
  • 批处理:设计支持批量操作的API以减少调用频率

贡献指南

欢迎对此框架进行贡献:

  1. 分叉仓库
  2. 创建功能分支(git checkout -b feature/amazing-feature
  3. 提交更改(git commit -m '添加惊人的功能'
  4. 推送到分支(git push origin feature/amazing-feature
  5. 创建Pull Request

许可证

MIT许可证

参考资料