MCP SSE Client Python

MCP Playground

リモートの Model Context Protocol（MCP）エンドポイントと連携するための Python ツールキットです。Streamable HTTP（デフォルト）および SSE トランスポートに対応し、複数のプロバイダーを通じて LLM がツールを選択できます。

クイックスタート

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

1 回限りの使用

接続を維持する必要のないシンプルなスクリプトの場合：

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

LLM によるツール選択

自然言語に基づいて、LLM に適切な 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）
• チャットモード：自動、チャットのみ、ツールのみ
• ツールの検出と呼び出し
• 接続ステータスのリアルタイム表示

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