pymcp-sse: Python MCP over SSE 库

一个轻量级且灵活的Python应用程序Model Context Protocol (MCP)实现，专注于健壮的HTTP/SSE传输。

特性

• 模块化框架：干净地实现了BaseMCPServer、BaseMCPClient和MultiMCPClient。
• HTTP/SSE传输：具有自动会话管理、可配置超时和重连处理的健壮HTTP/SSE实现。
• 并发任务执行：通过BaseMCPServer.run_with_tasks()方法轻松运行带有持久后台异步任务的服务器。
• 工具注册与发现：简单的装饰器式工具注册（@server.register_tool()）和标准的describe_tools端点供客户端动态查询详细的工具能力（参数、描述）。
• 服务器推送：内置支持服务器发起的推送通知到客户端和定期存活心跳。包括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("我的简单服务器")

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

# 使用标准方法运行服务器
if __name__ == "__main__":
    # 其他参数传递给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("我的后台任务服务器")

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

# 定义关闭回调
async def cleanup():
    print("清理中...")

# 使用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"已连接。可用工具: {client.available_tools}")
            # 调用工具
            result = await client.call_tool("echo", text="你好，世界！")
            print(f"工具结果: {result}")
            # 如需分配通知处理器
            # client.notification_handler = your_async_handler
    except Exception as e:
        print(f"发生错误: {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}")
        
        # 获取有关已连接服务器的信息（包括工具详情）
        server_info = client.get_server_info()
        print("\n服务器信息:")
        for name, info in server_info.items():
             print(f"- {name}: 状态={info['status']}, 工具数={len(info.get('available_tools', []))}, 详情获取={bool(info.get('tool_details'))}")

        # 在特定服务器上调用工具
        if server_info.get("server_basic", {}).get("status") == "connected":
            result = await client.call_tool("server_basic", "echo", text="来自MultiClient的问候！")
            print(f"\n基本服务器回显结果: {result}")
    except Exception as e:
        print(f"发生错误: {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