AI导哈

FastAPI转MCP服务器工具

mcp-it是一个将FastAPI服务无缝转换为MCP（Model Context Protocol）服务器的工具，通过简单的装饰器即可实现API功能的转换与扩展。

  • by enzobjmendonca
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 enzobjmendonca/mcp-it

使用说明

mcp它

将任何FastAPI服务器转换为MCP（模型上下文协议）服务器。

mcp允许您将FastAPI路由无缝转换为mcp工具。用以下方式装饰您的FastAPI端点 @mcp_it.mcp() 您的API将成为MCP服务器，可与兼容MCP的客户端一起使用。

特性

  • 最少的代码更改：使用简单的装饰器将现有的FastAPI路由转换为MCP工具
  • 自动参数检测：自动检测FastAPI路由中的路径、查询和正文参数-无需样板重新实现
  • 标头转发：将MCP请求的标头映射并保留到FastAPI端点
  • 依赖注入：完全支持FastAPI的依赖注入系统
  • Pydantic模型：与Pydantic模型完全集成，用于请求/响应验证
  • 多个传输：支持sse和可流式传输的http
  • 类型安全：维护类型提示和注释，以进行正确的MCP工具自检
  • 自动文档：使用您的函数docstring自动记录MCP工具
  • 代理支持：代理外部API作为MCP工具，具有自动参数映射功能
  • OpenAPI绑定：动态注册任何OpenAPI规范中的工具-立即将任何REST API转换为MCP工具

安装

pip install mcp-it

快速开始

from fastapi import FastAPI, APIRouter
from mcpit import MCPIt

# Create your FastAPI app
app = FastAPI()

# Initialize MCPIt
mcp_it = MCPIt("My API Tools", json_response=True)

# Decorate your routes with @mcp_it.mcp()
@app.post("/multiply")
@mcp_it.mcp(mode='tool', description="Multiply two numbers")
def multiply(a: float, b: float):
    """Multiplies two numbers together."""
    return {"result": a * b}

# Build and mount the MCP server
mcp_it.build(app.router, transport='streamable-http', mount_path="/mcp")

# Your MCP server is now available at /mcp

用法

基础示例

from fastapi import FastAPI, APIRouter
from mcpit import MCPIt
from pydantic import BaseModel, Field

app = FastAPI()
router = APIRouter()
mcp_it = MCPIt("Calculator Tools", json_response=True)

class CalculationInput(BaseModel):
    a: float = Field(..., description="The first number")
    b: float = Field(..., description="The second number")

@router.post("/add")
@mcp_it.mcp(mode='tool', name="add", description="Add two numbers") #If no route is annonotaded MCP It will make all routes on the router available
def add(input: CalculationInput):
    """Adds two numbers together."""
    return {"result": input.a + input.b}

mcp_it.build(router, transport='streamable-http', mount_path="/mcp")
app.include_router(router)

通过身份验证

mcp-it 自动转发来自MCP请求的标头，因此您的身份验证依赖关系可以无缝工作：

from fastapi import Header, Depends
from typing import Optional

def fake_auth_dependency(authorization: Optional[str] = Header(None)):
    """Validate authorization header"""
    if authorization:
        return True
    return False

@router.get("/protected")
@mcp_it.mcp(mode='tool', description="Protected endpoint")
def protected_route(auth_result: bool = Depends(fake_auth_dependency)):
    """A protected route that requires authentication."""
    return {"message": "Access granted", "authenticated": auth_result}

通过依赖注入

完全支持FastAPI的依赖注入：

class BasicService:
    def __init__(self):
        self.counter = 0
    
    def increment(self) -> int:
        self.counter += 1
        return self.counter

service = BasicService()
def get_service() -> BasicService:
    return service

from typing import Annotated
from fastapi import Depends

ServiceDep = Annotated[BasicService, Depends(get_service)]

@router.get("/increment")
@mcp_it.mcp(mode='tool', description="Increment counter")
def increment(service: ServiceDep):
    """Increment the counter using dependency injection."""
    return {"count": service.increment()}

路径和查询参数

mcp-it 自动检测和处理路径和查询参数：

@router.get("/users/{user_id}/posts")
@mcp_it.mcp(mode='tool', description="Get user posts")
def get_user_posts(user_id: int, limit: int = 10, offset: int = 0):
    """Get posts for a specific user with pagination."""
    return {
        "user_id": user_id,
        "posts": [],
        "limit": limit,
        "offset": offset
    }

代理外部API

将任何外部API端点转换为MCP工具。非常适合为多种服务创建统一的MCP接口：

from fastapi import FastAPI, APIRouter
from mcpit import MCPIt

app = FastAPI()
router = APIRouter()
mcp_it = MCPIt("Proxy Tools", json_response=True)

# Proxy an external API endpoint
@mcp_it.proxy(
    url="http://api.example.com/multiply", 
    method="POST", 
    param_structure={"a": "query", "b": "query"}
)
def multiply(a: float, b: float):
    """Multiply two numbers via external API"""
    pass

mcp_it.build(router, transport='streamable-http', mount_path="/mcp")
app.include_router(router)

这 param_structure 参数告诉 mcp-it 其中应发送每个参数：

  • "query"：查询参数
  • "path"：路径参数（将替换 {param_name} 在URL中）
  • "body"：请求正文参数

OpenAPI绑定

根据任何OpenAPI规范自动生成MCP工具。这非常适合：

  • 将现有的REST API打包为MCP服务器
  • 创建与第三方服务的MCP接口
  • 根据API文档快速原型化MCP工具
from fastapi import FastAPI, APIRouter
from mcpit import MCPIt

app = FastAPI()
router = APIRouter()
mcp_it = MCPIt("API Tools", json_response=True)

# Bind tools from an OpenAPI specification
mcp_it.bind_openapi(
    openapi_url="http://localhost:8800/openapi.json",
    include_paths=["/hello", "/multiply", "/subtract_with_auth"],  # Optional: filter specific paths
    exclude_paths=["/admin"],  # Optional: exclude paths
    name_from_summary=False  # Use operationId (True) or summary (False) for tool names
)

# You can also combine with manual tools
@router.post("/custom")
@mcp_it.mcp(mode='tool', description="Custom tool")
def custom_tool():
    """A custom tool alongside OpenAPI-bound tools"""
    return {"message": "Hello"}

mcp_it.build(router, transport='streamable-http', mount_path="/mcp")
app.include_router(router)

OpenAPI绑定特性：

  • 自动解析OpenAPI 3.x规范
  • 提取路径、查询和正文参数
  • 从模式生成正确的类型注释
  • 支持过滤 include_paths 和 exclude_paths
  • 保留OpenAPI规范中的描述和摘要

api参考

MCPIt

从FastAPI路由创建MCP服务器的主要类。

构造函数

MCPIt(name: str, json_response: bool = True)
  • 名字：MCP服务器的名称
  • jsonresponse：是否返回JSON响应（默认值：True）

方法

mcp(mode='tool', **kwargs)

Decorator将FastAPI路由注册为MCP功能。

参数：

  • 模式：MCP能力类型('tool', 'resource'，或 'prompt')
  • 名字：（可选）工具的自定义名称（默认为函数名）
  • 描述：（可选）工具描述（默认为函数docstring）

例子：

@mcp_it.mcp(mode='tool', name="custom_name", description="Custom description")
def my_function():
    pass
proxy(url, method='GET', mode='tool', **kwargs)

Decorator将外部API端点注册为MCP功能。装饰函数的签名定义了工具接口，而实现则代理到指定的URL。

参数：

  • 网址：外部API终结点URL
  • 方法：HTTP方法('GET', 'POST', 'PUT', 'DELETE'等等）
  • 模式：MCP能力类型('tool', 'resource'，或 'prompt')
  • 参数结构：（可选）Dict将参数名称映射到其位置('query', 'path'，或 'body')
  • 名字：（可选）工具的自定义名称（默认为函数名）
  • 描述：（可选）工具描述（默认为函数docstring）

例子：

@mcp_it.proxy(
    url="http://api.example.com/users/{user_id}",
    method="GET",
    param_structure={"user_id": "path", "include_posts": "query"}
)
def get_user(user_id: int, include_posts: bool = False):
    """Get user information from external API"""
    pass
bind_openapi(openapi_url, base_url=None, include_paths=None, exclude_paths=None, name_from_summary=False)

根据OpenAPI规范动态注册工具。

参数：

  • openapi_url：OpenAPI JSON规范的URL
  • base_url：（可选）API调用的基本URL。如果未提供，则从以下内容推断 openapi_url
  • include_paths：（可选）要包括的路径模式列表（例如。， ["/users", "/posts"])
  • 排除路径：（可选）要排除的路径模式列表（例如。， ["/admin"])
  • name_from_summary：（可选）如果 True，使用操作摘要作为工具名称；如果 False，使用 operationId （默认值： False)

例子：

mcp_it.bind_openapi(
    openapi_url="https://api.example.com/openapi.json",
    include_paths=["/users", "/posts"],
    exclude_paths=["/admin"],
    name_from_summary=False
)
build(router, transport='streamable-http', mount_path='/mcp')

构建MCP服务器并将其安装到FastAPI路由器。

参数：

  • 路由器：您的FastAPI APIRouter 例子
  • 运输：运输类型('sse' 或 'streamable-http')
  • mount_path：装载MCP服务器的路径（默认： /mcp)

例子：

mcp_it.build(router, transport='streamable-http', mount_path="/mcp")

运作原理

本地FastAPI路由

  1. 装修商注册：使用时 @mcp_it.mcp()，该函数已在内部注册表中注册
  2. 路径分析：在 build(), mcp-it 分析您的FastAPI路由以检测：
    • 路径参数
    • 查询参数
    • 车身参数
    • Pydantic模型
  3. 工具创建：MCP工具是使用适当的签名和类型注释创建的
  4. 内部代理：当调用MCP工具时， mcp-it 对FastAPI路由进行内部ASGI调用
  5. 标头转发：MCP请求的标头会自动转发到您的端点

外部API代理

  1. 代理注册：使用时 @mcp_it.proxy()，函数签名定义了工具接口
  2. 参数映射：The param_structure 参数告诉 mcp-it 在哪里发送每个参数（查询、路径或正文）
  3. 外部呼叫：当调用工具时， mcp-it 向外部API发出HTTP请求
  4. 标头转发：MCP请求的标头会自动转发到外部API

OpenAPI绑定

  1. 规格获取: bind_openapi() 从提供的URL获取OpenAPI规范
  2. 路径筛选：可选 include_paths 和 exclude_paths 筛选要公开的端点
  3. 架构解析OpenAPI模式被解析并转换为Python类型和Pydantic模型
  4. 工具生成：MCP工具会自动生成，并带有适当的签名和描述
  5. 代理创建：每个工具都注册为调用原始API端点的代理

运输方式

流式HTTP（推荐）

mcp_it.build(router, transport='streamable-http', mount_path="/mcp")
  • 使用支持流媒体的HTTP
  • 更适合生产环境
  • 支持所有HTTP方法

SSE（服务器发送事件）

mcp_it.build(router, transport='sse', mount_path="/mcp")
  • 使用服务器发送的事件进行流式传输
  • 适合实时更新
  • 更简单的连接模型

发展

设置

# Clone the repository
git clone https://github.com/enzobjmendonca/mcp-it.git
cd mcp-it

# Install development dependencies
pip install -r requirements.txt

# Install the package in editable mode
pip install -e .

运行测试

pytest

示例

基本示例(example.py)

请参阅 example.py 存储库中的文件，以获取完整的工作示例，包括：

  • 多种路线类型
  • Pydantic模型
  • 认证
  • 依赖注入
  • 路径和查询参数

代理示例(example_proxy.py)

请参阅 example_proxy.py 文件示例：

  • 手动配置代理 @mcp_it.proxy()
  • OpenAPI绑定 bind_openapi()
  • 将代理工具与本地FastAPI路由相结合

需求

  • Python 3.8+
  • 快速API
  • FastMCP
  • HTTPX
  • 派丹蒂克

贡献

我们非常欢迎您的贡献！请随时提交拉取请求。

许可证

MIT许可证-请参阅 许可证 文件以获取详细信息。

链接

相关 MCP