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助手类。 - LLM集成：包括
BaseLLMClient抽象，便于与各种LLM提供商集成（提供了一个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实现说明、LLM集成指南和协议规范，请参阅 docs/ 目录。
例子
请参阅 examples/ 完整工作示例目录，包括：
server_basic：演示一个简单的服务器，使用server.run().server_tasks：使用演示具有后台任务（通知调度程序）的服务器server.run_with_tasks().client：使用的多服务器客户端MultiMCPClient和aLLMAgent通过自然语言与两台服务器交互。需要API密钥（集ANTHROPIC_API_KEY在一个.env项目根目录中的文件）。run_all.py：一个易于启动的启动器脚本server_basic,server_tasks，以及client同时。notification_listener.py：一个简单的独立客户端，用于接收来自任何兼容服务器的推送通知。
许可证
麻省理工学院