Streamable HTTPに対応したMCP Python SDKを試してみる
本記事では、Streamable HTTP Transportに対応したMCP Python SDK v1.8.0とlangchain-mcp-adapters v0.0.11を使い、ステートレスかつSSEを用いずにtool callingをするAIエージェントを実装します。
実装するもの
今回は、以下の機能を持つサーバーとクライアントを実装します。
- サーバー側：MCPで数値計算(加算、乗算)を行うToolsを提供
- クライアント側：langchain-mcp-adaptersとLangChain/LangGraphとを使ってMCPサーバ上のtoolをcallし、ユーザの要求に応えるAIエージェント
必要なパッケージ
以下のコマンドで必要なライブラリをインストールします
サーバーに必要なライブラリ
mcpのPython SDKをインストールします。これにより、MCPを使用したサーバーを実装できます。
pip install 'mcp[cli]>=1.8.0'
クライアントに必要なライブラリ
AIエージェントを実装するために必要なパッケージ(langgraph)やlangchain-mcp-adaptersをインストールします。langchain-mcp-adaptersは、MCP介してtoolを呼び出すためのアダプターです。
pip install 'langchain-mcp-adapters>=0.0.11' langgraph langchain-openai
環境変数の設定
AI Agentを動かすためのLLMとして、gpt-4.1-nanoを使用します。
OpenAIのAPIにアクセスするために環境変数OPENAI_API_KEYを設定します。
export OPENAI_API_KEY=your_api_key_here
サーバー側の実装
MCPサーバーは数値計算のための2つのツールを提供します
from mcp.server.fastmcp import FastMCP
# stateless_htte=Trueとすると、ステートレスな通信になります
# json_response=Trueとすると、サーバからのレスポンスがSSEでなくなります
# log_levelをDEBUGにすると、通信やレスポンスの状態を把握しやすくなります
mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True, log_level="DEBUG")
@mcp.tool('add')
def add(a: int, b: int) -> int:
"""
Add two integers.
"""
return a + b
@mcp.tool('multiply')
def multiply(a: int, b: int) -> int:
"""
Multiply two integers.
"""
return a * b
# Run server with streamable_http transport
if __name__ == "__main__":
mcp.run(transport="streamable-http")
クライアント側の実装
LangChainとLangGraphを使ってMCPサーバーのツールを利用するクライアントを実装します
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from langgraph.prebuilt import create_react_agent
from langchain_mcp_adapters.tools import load_mcp_tools
async def main():
async with streamablehttp_client("http://localhost:8000/mcp") as (read, write, _):
async with ClientSession(read, write) as session:
# セッションを初期化
await session.initialize()
# MCPからtools情報を取得
tools = await load_mcp_tools(session)
# AIエージェントのセットアップ
agent = create_react_agent("openai:gpt-4.1-nano", tools)
# メッセージを送信
response = await agent.ainvoke({"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]})
print(response)
if __name__ == "__main__":
asyncio.run(main())
実行方法
- まずサーバーを起動します：
python server.py
- 別のターミナルでクライアントを実行します：
python client.py
動作の流れ
- MCPサーバーは加算と乗算の機能をMCP Toolとして公開します
- クライアントはサーバーに接続し、利用可能なツール情報を取得します
- LangGraphによって作成されたエージェントがユーザーの質問を理解し、必要なツールを呼び出します
- 例えば「(3 + 5) x 12は？」という質問に対しては：
- エージェントはまず
addツールを使って3 + 5 = 8を計算 - 次に
multiplyツールを使って8 x 12 = 96を計算 - 最終的に「96です」と回答します
- エージェントはまず
まとめ
MCPを使うことで、異なるシステム間でAIツールを簡単に連携させることができます。
これまで、MCPはstdioをやSSEを用いた実行が主流でしたが、今回のようにStreamable HTTPを利用することで、ステートレスな通信やSSEを用いないリモートのMCPが実装可能になります。
特にSSEは対応していない環境（AWS API Gatewayなど）が存在し、MCPの採用の壁となっていましたが、その壁がなくなったことで、今後様々な環境での利用が広まることでしょう。
参考リンク
MCP Python SDK
langchain-mcp-adapters
MCP TypeScript SDK
Discussion