pymcp-sse: Python MCP over SSE 库

一个轻量级、灵活的 Python 应用程序模型上下文协议 (MCP) 实现，专门用于健壮的 HTTP/SSE 传输。

特性

• 模块化框架: BaseMCPServer、BaseMCPClient 和 MultiMCPClient 的清晰实现。
• HTTP/SSE 传输: 健壮的 HTTP/SSE 实现，具有自动会话管理、可配置的超时和重新连接处理。
• 并发任务执行: BaseMCPServer.run_with_tasks() 方法，用于轻松运行具有持久后台异步任务的服务器。
• 工具注册与发现: 简单的基于装饰器的工具注册 (@server.register_tool()) 和一个标准的 describe_tools 端点，供客户端动态查询详细的工具功能（参数、描述）。
• 服务器推送: 内置支持服务器发起的向客户端的推送通知和周期性保持活动 ping。包括 NotificationScheduler 辅助类。
• LLM 集成: 包括 BaseLLMClient 抽象，以便轻松与各种 LLM 提供商集成（提供了一个 Anthropic Claude 示例）。
• 灵活的日志记录: 可通过 pymcp_sse.utils 配置日志记录。

安装

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

# 导航到包含 pyproject.toml 的目录
cd /path/to/your/pymcp-sse

# 以可编辑模式安装
pip install -e .

（一旦发布，将可以通过 pip install pymcp-sse 进行安装。）

基本用法

创建一个 MCP 服务器（简单）

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

configure_logging() # 配置日志记录（可选）

# 创建一个服务器实例
server = BaseMCPServer("My Simple Server")

# 使用装饰器注册工具
# 类型提示被 describe_tools 使用
@server.register_tool("echo")
async def echo_tool(text: str) -> dict:
    '''回显提供的文本。'''
    return {"response": f"Echo: {text}"}

# 使用标准方法运行服务器
if __name__ == "__main__":
    # 附加的 kwargs 传递给 uvicorn.run (例如，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() # 配置日志记录（可选）

# 创建一个服务器实例
server = BaseMCPServer("My Background Task Server")

# 定义你的后台任务
async def my_periodic_task():
    while True:
        print("Task running...")
        await asyncio.sleep(5)

# 定义一个关闭回调
async def cleanup():
    print("Cleaning up...")

# 使用 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() # 配置日志记录（可选）

async def main():
    # 配置超时以提高稳定性（读取超时 > 服务器 ping 间隔）
    client = BaseMCPClient(
        "http://localhost:8000", # 指向你的服务器
        http_read_timeout=65, 
        http_connect_timeout=10
    )
    
    try:
        # 连接并初始化
        if await client.connect() and await client.initialize():
            print(f"Connected. Available tools: {client.available_tools}")
            # 调用一个工具
            result = await client.call_tool("echo", text="Hello, world!")
            print(f"Tool Result: {result}")
            # 如果需要，分配一个通知处理程序
            # 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() # 配置日志记录（可选）

async def main():
    # 使用示例部分中的服务器
    servers = {
        "server_basic": "http://localhost:8101",
        "server_tasks": "http://localhost:8102"
    }
    # 配置超时以提高稳定性（读取超时 > 服务器 ping 间隔）
    client = MultiMCPClient(
        servers,
        http_read_timeout=65,
        http_connect_timeout=10
    )
    
    try:
        # 连接到所有服务器（如果存在 describe_tools，则自动获取工具详细信息）
        connection_results = await client.connect_all()
        print(f"Connection Results: {connection_results}")
        
        # 获取有关已连接服务器的信息（包括工具详细信息）
        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'))}")

        # 在特定服务器上调用一个工具
        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 实现的说明、LLM 集成指南和协议规范，请参阅 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