LiteLLMとLangGraphでMCPサーバーのツールをエージェントから利用してみる

Python 3.13
litellm 1.83.14
langgraph 1.1.10
langchain-mcp-adapters 0.1.0
mcp 1.5.0
langchain-litellm 0.6.4

$ uv pip install litellm langgraph langchain-litellm langchain-mcp-adapters mcp

"""MCP サーバー側の最小サンプル: 開発者環境の情報を返すツール群。

stdio トランスポートで動く。クライアントから
python sample_mcp_server.py で起動される。
"""

import sys
from importlib.metadata import distributions

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("dev-env-tools")

@mcp.tool()
def get_python_version() -> str:
    """現在実行中の Python のバージョン情報を返します。"""
    return sys.version

@mcp.tool()
def list_installed_packages(prefix: str = "") -> list[str]:
    """インストール済み Python パッケージを返します。

    prefix を指定すると、その文字列で始まるパッケージのみを返します
    （例: prefix="fastapi" で fastapi 系だけに絞り込み）。
    """
    packages = sorted(
        f"{dist.metadata['Name']}=={dist.version}"
        for dist in distributions()
        if dist.metadata["Name"]
    )
    if prefix:
        packages = [p for p in packages if p.lower().startswith(prefix.lower())]
    return packages

if __name__ == "__main__":
    mcp.run(transport="stdio")

"""LangGraph エージェントから MCP サーバーのツールを使う。

langchain-mcp-adapters の MultiServerMCPClient で
- ローカル sample_mcp_server.py（stdio）
- リモート Context7 MCP サーバー（streamable_http）
の2系統を束ね、1つのエージェントから利用する。
"""

import asyncio
import sys

from langchain.agents import create_agent
from langchain_litellm import ChatLiteLLM
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main() -> None:
    # 複数のMCPサーバーを束ねる
    client = MultiServerMCPClient(
        {
            # ローカルのstdioサーバー
            "sample": {
                "command": sys.executable,
                "args": ["sample_mcp_server.py"],
                "transport": "stdio",
            },
            # リモートの Streamable HTTP サーバー（Context7: ライブラリのドキュメント検索）
            "context7": {
                "url": "https://mcp.context7.com/mcp",
                "transport": "streamable_http",
            },
        }
    )

    # MCP サーバーが提供するツール群を取得（自動的に LangChain Tool に変換）
    tools = await client.get_tools()
    print(f"取得したツール: {[t.name for t in tools]}")

    llm = ChatLiteLLM(model="openai/gpt-5-mini")
    agent = create_agent(model=llm, tools=tools)

    # 「ローカル環境を確認 → そのバージョンに合ったドキュメントを Context7 で引く」
    # という自然なワークフローを1つのエージェントで完結させる。
    result = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "このプロジェクトにインストールされている fastapi のバージョンを確認し、"
                        "そのバージョンに合わせて StreamingResponse の使い方を Context7 で調べ、"
                        "ポイントを3行で要約してください。"
                    ),
                }
            ]
        }
    )
    print("\n=== 最終回答 ===")
    print(result["messages"][-1].content)

asyncio.run(main())

$ python mcp_client.py
INFO     Processing request of type ListToolsRequest      server.py:727
INFO     Processing request of type CallToolRequest       server.py:727
INFO     Processing request of type ListToolsRequest      server.py:727
取得したツール: ['get_python_version', 'list_installed_packages', 'resolve-library-id', 'query-docs']

=== 最終回答 ===
fastapi==0.136.1 がインストールされています。
StreamingResponse は async generator または通常の generator/iterator を content に取り、bytes（または文字列）チャンクを逐次送信でき、media_type / status_code / headers を指定して大きなレスポンスをメモリ効率よく配信します。
ストリーム後のクリーンアップは BackgroundTasks（または StreamingResponse の background 引数）で行い、プロキシ時は hop-by-hop ヘッダを除外、キャンセル対応のために await anyio.sleep(0) 等のチェックポイントを入れるのが推奨されます。