FastAPI MCP Client

FastAPI MCP 客户端

一个专用的客户端库，专为与实现了基于服务器发送事件 (Server-Sent Events, SSE) 的模型上下文协议 (Model Context Protocol, MCP) 的 fastapi-mcp 服务配合使用而设计。该库提供了与启用 MCP 的 FastAPI 服务交互的无缝方式。

安装

# Install with pip
pip install fastapi-mcp-client

# Or with UV
uv add fastapi-mcp-client

快速开始

import asyncio
from fastapi_mcp_client import MCPClient

async def main():
    async with MCPClient("http://localhost:8000") as client:
        # Call a non-streaming operation
        result = await client.call_operation("echo", {"message": "Hello, MCP!"})
        print(f"Echo result: {result}")
        
        # Call a streaming operation with SSE
        stream = await client.call_operation(
            "generate_numbers", 
            {"count": 5},
            stream=True
        )
        
        async for event in stream:
            print(f"Event: {event}")

asyncio.run(main())

功能特性

• MCP 协议支持：模型上下文协议的完整实现
• SSE 流式传输：对服务器发送事件 (SSE) 流式传输的一流支持
• 异步优先设计：为高性能应用提供完整的异步兼容性
• 无缝会话管理：处理 MCP 会话建立和消息传递
• 错误处理：全面的错误处理和回退机制
• 类型注解：完整的类型提示，可更好地集成 IDE 并进行验证

运行示例

该仓库包含示例以帮助你快速开始。以下是运行示例的方式：

1. 克隆仓库

git clone https://github.com/RooseveltAdvisors/fastapi-mcp-client.git
cd fastapi-mcp-client

2. 设置环境

# Create and activate virtual environment
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install all dependencies including examples
uv sync --all-extras

3. 启动示例服务

# In one terminal
cd examples/server
python simpler_server.py

这将启动一个 FastAPI 服务，其中包含多个启用 MCP 的端点：

• echo：返回你发送的消息
• generate_numbers：流式传输数字序列
• search_documents：模拟带有流式结果的文档搜索
• calculate：计算一个简单的数学表达式

4. 运行 SSE 客户端示例

# In another terminal (with virtual env activated)
cd examples
python sse_example.py

高级用法

自定义客户端配置

from fastapi_mcp_client import MCPClient, MCPClientConfig

config = MCPClientConfig(
    base_url="http://localhost:8000",
    timeout=60.0,
    log_level="DEBUG"
)

async with MCPClient("http://localhost:8000", config=config) as client:
    # Call a streaming tool with custom configuration
    stream = await client.call_operation(
        "generate_numbers", 
        {"count": 10},
        stream=True
    )
    
    async for event in stream:
        print(f"Received event: {event}")

理解 MCP/SSE 流程

sequenceDiagram
    participant C as Client
    participant S as Server

    Note over C,S: Establish SSE Connection
    C->>+S: GET /mcp (Accept: text/event-stream)
    S-->>-C: 200 OK (Connection Open)
    S-->>C: SSE: data: /mcp/messages/?session_id=XXX

    Note over C: Parse session_id=XXX

    Note over C,S: MCP Initialization
    C->>+S: POST /mcp/messages/?session_id=XXX <br> Payload: {method: "initialize", ...}
    S-->>-C: 202 Accepted

    Note over C,S: MCP Tool Call
    C->>+S: POST /mcp/messages/?session_id=XXX <br> Payload: {method: "tools/call", ...}
    S-->>-C: 202 Accepted

    Note over C,S: Stream Results
    S-->>C: SSE: data: {result_part_1}
    S-->>C: SSE: data: {result_part_2}
    S-->>C: SSE: data: {final_result}
    Note over S: (Closes SSE Connection or sends close event)

示例

查看 examples 目录获取更多内容：

• 服务示例：示例 FastAPI MCP 服务
• SSE 示例：具有自定义配置的高级流式传输

贡献

欢迎贡献！请随时提交 Pull Request。

开发设置

1. 克隆仓库

   git clone https://github.com/RooseveltAdvisors/fastapi-mcp-client.git
   cd fastapi-mcp-client
   

2. 安装 uv

   curl -LsSf https://astral.sh/uv/install.sh | sh
   

3. 创建并激活虚拟环境

   uv venv
   source .venv/bin/activate  # On Windows: .venv\Scripts\activate
   

4. 安装开发依赖

   uv sync --all-extras
   

5. 运行测试

   pytest
   

Pull Request 流程

1. 创建你的功能分支 (git checkout -b feature/amazing-feature)
2. 进行更改并确保测试通过
3. 提交你的更改 (git commit -m 'Add some amazing feature')
4. 推送到分支 (git push origin feature/amazing-feature)
5. 开启 Pull Request

许可证

该项目采用 MIT 许可证授权 - 详见 LICENSE 文件。

致谢

• 为与 FastAPI MCP 配合使用而构建 - 这是一个为你的 API 添加 MCP 支持的 FastAPI 扩展