pymcp-sse: Python MCP over SSE Library

- Modular Framework: Clean implementation of BaseMCPServer, BaseMCPClient, and MultiMCPClient.
- HTTP/SSE Transport: Robust HTTP/SSE implementation with automatic session management, configurable timeouts, and reconnection handling.
- Concurrent Task Execution: BaseMCPServer.run_with_tasks() method for easily running servers with persistent background asynchronous tasks.
- Tool Registration & Discovery: Simple decorator-based tool registration (@server.register_tool()) and a standard describe_tools endpoint for clients to dynamically query detailed tool capabilities (parameters, descriptions).
- Server Push: Built-in support for server-initiated push notifications to clients and periodic keep-alive pings. Includes NotificationScheduler helper class.
- LLM Integration: Includes BaseLLMClient abstraction for easy integration with various LLM providers (an Anthropic Claude example is provided).
- Flexible Logging: Configurable logging via pymcp_sse.utils.

To install the library locally for development:

pip install -e .

(Once published, installation via pip install pymcp-sse will be available.)

client = BaseMCPClient(
"http://localhost:8000", # Point to your server
http_read_timeout=65,
http_connect_timeout=10
)

try:

@server.register_tool("echo")
async def echo_tool(text: str) -> dict:
'''Echoes the provided text back.'''
return {"response": f"Echo: {text}"}

result = await client.call_tool("echo", text="Hello, world!")
print(f"Tool Result: {result}")

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())
```