Fastapi MCP Client

FastAPI MCP客户端

专门设计用于与 fastapi-mcp 通过服务器发送事件（SSE）实现模型上下文协议（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

这将启动一个具有多个启用MCP的端点的FastAPI服务器：

• 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 
 Payload: {method: "initialize", ...}
    S-->>-C: 202 Accepted

    Note over C,S: MCP Tool Call
    C->>+S: POST /mcp/messages/?session_id=XXX 
 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)

例子

看看 示例 更多目录：

• 服务器示例：FastAPI MCP服务器示例
• SSE示例：具有自定义配置的高级流媒体

贡献

欢迎投稿！请随时提交拉取请求。

开发设置

1. 克隆存储库

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

1. 安装 紫外线

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

1. 创建并激活虚拟环境

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

1. 安装开发依赖项

   uv sync --all-extras

1. 运行测试

   pytest

拉取请求流程

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

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。

致谢

• 为配合使用而构建 FastAPI MCP -FastAPI扩展，用于向您的API添加MCP支持