MCP SSE Client Python

MCP Playground

一个用于与远程 Model Context Protocol (MCP) 端点交互的 Python 工具包。支持 Streamable HTTP（默认）和 SSE 传输，并可在多个提供商之间进行由大模型驱动的工具选择。

快速开始

git clone https://github.com/zanetworker/mcp-playground.git
cd mcp-playground
pip install -e ".[all]"

用法

上下文管理器（推荐）

import asyncio
from mcp_playground import MCPClient

async def main():
    async with MCPClient("http://localhost:8000/mcp") as client:
        tools = await client.list_tools()
        print(f"Found {len(tools)} tools")

        result = await client.call_tool("add", arguments={"a": 10, "b": 5})
        print(f"Result: {result.content}")

asyncio.run(main())

SSE 传输（旧版服务）

async with MCPClient("http://localhost:8000/sse", transport="sse") as client:
    tools = await client.list_tools()

自定义认证 / 请求头

import httpx

http = httpx.AsyncClient(headers={"Authorization": "Bearer token"})
async with MCPClient("http://localhost:8000/mcp", http_client=http) as client:
    tools = await client.list_tools()

一次性用法

适用于不需要持久连接的简单脚本：

client = MCPClient("http://localhost:8000/mcp")
tools = await client.list_tools()  # auto-connects and disconnects

由大模型驱动的工具选择

让大模型根据自然语言选择并执行正确的 MCP 工具：

import os
from mcp_playground import MCPClient
from mcp_playground.llm_bridge import OpenAIBridge

async with MCPClient("http://localhost:8000/mcp") as client:
    bridge = OpenAIBridge(client, api_key=os.environ["OPENAI_API_KEY"])
    result = await bridge.process_query("Add 10 and 5")

    if result["tool_call"]:
        print(f"Tool: {result['tool_call']['name']}")
        print(f"Result: {result['tool_result'].content}")

支持的 LLM 提供商

提供商	桥接类	默认模型
OpenAI	OpenAIBridge	gpt-4o
Anthropic	AnthropicBridge	claude-sonnet-4-20250514
Ollama	OllamaBridge	llama3.3
OpenRouter	OpenRouterBridge	任意模型 ID

交互式 Playground (Streamlit)

cd mcp-streamlit-app
pip install -r requirements.txt
streamlit run app.py

功能：

• 传输方式选择器（Streamable HTTP / SSE）
• 多提供商 LLM 支持（OpenAI、Anthropic、Ollama、OpenRouter）
• 聊天模式：Auto、Chat-only、Tools-only
• 工具发现与调用
• 实时连接状态

API 参考

MCPClient

client = MCPClient(
    endpoint="http://localhost:8000/mcp",
    transport="streamable-http",  # or "sse"
    timeout=30.0,
    http_client=None,  # optional httpx.AsyncClient
)

方法：

方法	描述
list_tools()	列出 MCP 端点可用的工具
call_tool(name, arguments={})	使用参数调用工具
invoke_tool(name, kwargs)	call_tool() 的向后兼容别名
check_connection()	检查端点是否可达
get_endpoint_info()	获取端点元数据
connect() / disconnect()	手动管理会话生命周期

ToolInvocationResult

result = await client.call_tool("add", arguments={"a": 1, "b": 2})
result.content             # Text content as string
result.error_code          # 0 for success, 1 for error
result.structured_content  # Structured output (dict), if available
result.raw_content         # Raw content blocks from the SDK

错误处理

from mcp_playground import MCPClient, MCPConnectionError, MCPTimeoutError

try:
    async with MCPClient("http://localhost:8000/mcp") as client:
        tools = await client.list_tools()
except MCPConnectionError as e:
    print(f"Connection failed: {e}")
except MCPTimeoutError as e:
    print(f"Timed out: {e}")

要求

• Python >= 3.10
• mcp >= 1.26.0
• httpx >= 0.28.0
• pydantic >= 2.0.0

可选（用于 LLM 集成）：

• openai >= 2.28.0
• anthropic >= 0.84.0
• ollama >= 0.6.0

开发

pip install -e ".[dev]"
pytest                # run tests
black .               # format code
ruff check .          # lint

许可证

MIT