pymcp-sse: Python MCP over SSE Library

pymcp-sse：基于 SSE 的 Python MCP 库

面向 Python 应用的轻量、灵活的模型上下文协议（MCP）实现，专注于稳健的 HTTP/SSE 传输。

功能

• 模块化框架：简洁地实现了 BaseMCPServer、BaseMCPClient 和 MultiMCPClient。
• HTTP/SSE 传输：稳健的 HTTP/SSE 实现，支持自动会话管理、可配置超时和重连处理。
• 并发任务执行：BaseMCPServer.run_with_tasks() 方法，可轻松运行带有持久后台异步任务的服务。
• 工具注册与发现：通过装饰器轻松注册工具（@server.register_tool()），并提供标准的 describe_tools 端点，供客户端动态查询工具的详细功能（参数、说明）。
• 服务端推送：内置支持服务端主动向客户端推送通知和定期发送保活 ping。包含 NotificationScheduler 辅助类。
• 大模型集成：包含 BaseLLMClient 抽象类，可轻松集成各种大模型提供商（提供了 Anthropic Claude 示例）。
• 灵活的日志记录：通过 pymcp_sse.utils 配置日志记录。

安装

在本地安装此库以进行开发：

# Navigate to the directory containing pyproject.toml
cd /path/to/your/pymcp-sse

# Install in editable mode
pip install -e .

（发布后，可通过 pip install pymcp-sse 安装。）

基本用法

创建 MCP 服务（简单方式）

from pymcp_sse.server import BaseMCPServer
from pymcp_sse.utils import configure_logging

configure_logging() # Configure logging (optional)

# Create a server instance
server = BaseMCPServer("My Simple Server")

# Register tools using the decorator
# Type hints are used by describe_tools
@server.register_tool("echo")
async def echo_tool(text: str) -> dict:
    '''Echoes the provided text back.'''
    return {"response": f"Echo: {text}"}

# Run the server using the standard method
if __name__ == "__main__":
    # Additional kwargs are passed to uvicorn.run (e.g., timeout_keep_alive=65)
    server.run(host="0.0.0.0", port=8000)

创建 MCP 服务（带后台任务）

import asyncio
from pymcp_sse.server import BaseMCPServer
from pymcp_sse.utils import configure_logging

configure_logging() # Configure logging (optional)

# Create a server instance
server = BaseMCPServer("My Background Task Server")

# Define your background task
async def my_periodic_task():
    while True:
        print("Task running...")
        await asyncio.sleep(5)

# Define a shutdown callback
async def cleanup():
    print("Cleaning up...")

# Run the server using run_with_tasks
async def main():
    await server.run_with_tasks(
        host="0.0.0.0", 
        port=8001,
        concurrent_tasks=[my_periodic_task],
        shutdown_callbacks=[cleanup]
    )

if __name__ == "__main__":
    asyncio.run(main())

创建单个客户端

import asyncio
from pymcp_sse.client import BaseMCPClient
from pymcp_sse.utils import configure_logging

configure_logging() # Configure logging (optional)

async def main():
    # Configure timeouts for stability (read timeout > server ping interval)
    client = BaseMCPClient(
        "http://localhost:8000", # Point to your server
        http_read_timeout=65, 
        http_connect_timeout=10
    )
    
    try:
        # Connect and initialize
        if await client.connect() and await client.initialize():
            print(f"Connected. Available tools: {client.available_tools}")
            # Call a tool
            result = await client.call_tool("echo", text="Hello, world!")
            print(f"Tool Result: {result}")
            # Assign a notification handler if needed
            # client.notification_handler = your_async_handler
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

创建多服务端客户端

import asyncio
from pymcp_sse.client import MultiMCPClient
from pymcp_sse.utils import configure_logging

configure_logging() # Configure logging (optional)

async def main():
    # Use servers from the examples section
    servers = {
        "server_basic": "http://localhost:8101",
        "server_tasks": "http://localhost:8102"
    }
    # Configure timeouts for stability (read timeout > server ping interval)
    client = MultiMCPClient(
        servers,
        http_read_timeout=65,
        http_connect_timeout=10
    )
    
    try:
        # Connect to all servers (automatically fetches tool details if describe_tools exists)
        connection_results = await client.connect_all()
        print(f"Connection Results: {connection_results}")
        
        # Get info about connected servers (including tool details)
        server_info = client.get_server_info()
        print("\nServer Info:")
        for name, info in server_info.items():
             print(f"- {name}: Status={info['status']}, Tools={len(info.get('available_tools', []))}, Details Fetched={bool(info.get('tool_details'))}")

        # Call a tool on a specific server
        if server_info.get("server_basic", {}).get("status") == "connected":
            result = await client.call_tool("server_basic", "echo", text="Hello from MultiClient!")
            print(f"\nServer Basic Echo Result: {result}")
    except Exception as e:
        print(f"An error occurred: {e}")    
    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

文档

如需了解更详细的用法说明、HTTP/SSE 实现须知、大模型集成指南和协议规范，请参阅 docs/ 目录中的文档。

示例

请参阅 examples/ 目录中的完整可运行示例，其中包括：

• server_basic：演示使用 server.run() 的简单服务。
• server_tasks：演示使用 server.run_with_tasks() 运行带后台任务（通知调度器）的服务。
• client：使用 MultiMCPClient 和 LLMAgent，通过自然语言与两个服务交互的多服务端客户端。需要 API 密钥（在项目根目录的 .env 文件中设置 ANTHROPIC_API_KEY）。
• run_all.py：启动脚本，可轻松同时启动 server_basic、server_tasks 和 client。
• notification_listener.py：简单的独立客户端，用于接收任何兼容服务推送的通知。

许可证

MIT