モデルコンテキストプロトコル（MCP）は、AIモデルが外部ツールやサービスとどのように相互作用するかを標準化することを目的としています。これは共通のインターフェースを定義し、異なるモデルとツールプロバイダーが効果的にコミュニケーションできるようにします。しかし、MCP準拠のツールをLangChainのような既存のAIフレームワークに直接統合するには、適応が必要です。
ここで、langchain-mcp-adaptersライブラリが役立ちます。これは重要な橋渡しとして機能し、MCPツールをLangChainとその強力なエージェントフレームワークであるLangGraphが理解し利用できる形式にシームレスに変換します。このライブラリは軽量なラッパーを提供し、開発者が自分のLangChainアプリケーション内でMCPツールの成長するエコシステムを活用できるようにします。
主な機能は次のとおりです：
- MCPツール変換： MCPツールを自動的にLangChain互換の
BaseToolオブジェクトに変換します。 - マルチサーバークライアント： 複数のMCPサーバーに同時に接続できる堅牢なクライアント（
MultiServerMCPClient）を提供し、さまざまなソースからツールを集約します。 - トランスポートの柔軟性： 標準入力/出力（
stdio）やサーバー送信イベント（sse）などの一般的なMCP通信トランスポートをサポートします。
このチュートリアルでは、MCPサーバーのセットアップ、アダプターライブラリを使用した接続、ロードしたツールのLangGraphエージェントへの統合について説明します。
最大の生産性で開発チームが協力できる統合されたオールインワンプラットフォームが欲しいですか？
ApiDogはあなたのすべての要望に応え、はるかに手頃な価格でPostmanを置き換えます！
MCPサーバーとは何ですか？どのように機能しますか？
実例に飛び込む前に、いくつかの核心的概念を理解することが重要です：
MCPサーバー：
- MCPサーバーは、AIモデルが呼び出すことができるツール（関数）を公開します。
mcpライブラリ（langchain-mcp-adaptersの依存関係）が、Pythonでこれらのサーバーを簡単に作成するためのFastMCPなどのツールを提供します。- ツールは、
@mcp.tool()デコレーターを使用して定義され、タイプヒントとドキュメント文字列から入力スキーマを自動的に推測します。 - サーバーは
@mcp.prompt()を使用してプロンプトを定義することもでき、構造化された会話のスタートや指示を提供します。 - サーバーは、トランスポートメカニズム（例：
mcp.run(transport="stdio")またはmcp.run(transport="sse")）を指定して実行されます。stdioは、標準入力/出力を介して通信するサブプロセスとしてサーバーを実行し、一方sseは通常、通信のためにシンプルなWebサーバーを実行します。
MCPクライアント（langchain-mcp-adapters）：
- クライアントの役割は、1つまたは複数のMCPサーバーに接続することです。
- 通信プロトコルの詳細（stdio, sse）を処理します。
- サーバーから利用可能なツールのリストとその定義（名前、説明、入力スキーマ）を取得します。
MultiServerMCPClientクラスは、特に複数のツールサーバーを扱う場合に接続を管理するための主要な方法です。
ツール変換：
- MCPツールは独自の定義形式を持っています。LangChainは自分の
BaseToolクラス構造を使用します。 langchain-mcp-adaptersライブラリは、アクティブなClientSessionを介してサーバーに接続し、MCPツールのリストを取得し、それぞれをLangChainのStructuredToolにラッピングする関数load_mcp_tools（langchain_mcp_adapters.toolsにあります）を提供します。- このラッパーは、LangChainエージェントがツールを使用することを決定したときに実際のMCPツール呼び出し（
session.call_tool）を呼び出し、応答を正しくフォーマットします。
プロンプト変換：
- ツールと同様に、MCPプロンプトは
load_mcp_prompt（langchain_mcp_adapters.promptsから）を使用して取得できます。 - この関数はMCPサーバーからプロンプト構造を取得し、会話を初期化またはガイドするのに適したLangChainの
HumanMessageまたはAIMessageオブジェクトのリストに変換します。
Langchain-mcp-adapterのインストール
まず、必要なパッケージをインストールします：
pip install langchain-mcp-adapters langgraph langchain-openai # お好きなLangChain LLM統合に変更してください
選択した言語モデルプロバイダーのAPIキーを設定するために、環境変数を設定する必要があります：
export OPENAI_API_KEY=<your_openai_api_key>
# または export ANTHROPIC_API_KEY=<...> など
langchain-mcp-adaptersで簡単なMCPサーバーを構築する
数学関数を提供するMCPサーバーとそれらの関数を使用するLangGraphエージェントのシンプルな例を構築しましょう。
ステップ1：MCPサーバーを作成する（math_server.py）
# math_server.py
from mcp.server.fastmcp import FastMCP
# 名称でMCPサーバーを初期化する
mcp = FastMCP("Math")
@mcp.tool()
def add(a: int, b: int) -> int:
"""2つの数を加算します"""
print(f"Executing add({a}, {b})") # サーバー側のログ
return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""2つの数を掛けます"""
print(f"Executing multiply({a}, {b})") # サーバー側のログ
return a * b
# プロンプト定義の例
@mcp.prompt()
def configure_assistant(skills: str) -> list[dict]:
"""指定されたスキルでアシスタントを構成します。"""
return [
{
"role": "assistant", # AIMessageに対応
"content": f"あなたは役に立つアシスタントです。次のスキルを持っています: {skills}. 常に1つのツールだけを使用してください。",
}
]
if __name__ == "__main__":
# stdioトランスポートを使用してサーバーを実行する
print("Starting Math MCP server via stdio...")
mcp.run(transport="stdio")
このコードをmath_server.pyとして保存してください。
ステップ2：クライアントとエージェントを作成する（client_app.py）
import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
# --- 重要：このパスを更新 ---
# math_server.pyファイルへの絶対パスを取得
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# ---
async def main():
model = ChatOpenAI(model="gpt-4o") # またはお好きなモデル
# math_server.pyスクリプトを実行するためのパラメータを構成
server_params = StdioServerParameters(
command="python", # 実行するコマンド
args=[math_server_script_path], # 引数（スクリプトパス）
# cwd=..., env=... # オプションの作業ディレクトリと環境変数
)
print("MCPサーバーに接続しています...")
# stdio_clientコンテキストマネージャを使用して接続を確立します
async with stdio_client(server_params) as (read, write):
# 読み取り/書き込みストリームを使用してClientSessionを作成
async with ClientSession(read, write) as session:
print("セッションを初期化しています...")
# サーバーとハンドシェイク
await session.initialize()
print("セッションが初期化されました。")
print("MCPツールを読み込んでいます...")
# MCPツールを取得し、LangChainツールに変換
tools = await load_mcp_tools(session)
print(f"読み込まれたツール: {[tool.name for tool in tools]}")
# モデルと読み込まれたツールを使用してLangGraph ReActエージェントを作成します
agent = create_react_agent(model, tools)
print("エージェントを呼び出しています...")
# エージェントを実行
inputs = {"messages": [("human", "what's (3 + 5) * 12?")]}
async for event in agent.astream_events(inputs, version="v1"):
print(event) # 監視のためにイベントをストリーム
# または、最終応答を直接取得
# final_response = await agent.ainvoke(inputs)
# print("エージェントの応答:", final_response['messages'][-1].content)
if __name__ == "__main__":
asyncio.run(main())
これをclient_app.pyとしてmath_server.pyと同じディレクトリに保存してください。
実行するには：
クライアントスクリプトを実行します：
python client_app.py
クライアントスクリプトは自動的にmath_server.pyをサブプロセスとして起動し、接続し、addとmultiplyツールを読み込み、これらのツールを介してMCPサーバーに呼び出すことで数学の問題を解決するためにLangGraphエージェントを使用します。クライアントとサーバーの両方からのログが表示されます。
複数のMCPサーバーへの接続
異なる専門サーバーからのツールを組み合わせたい場合がよくあります。MultiServerMCPClientはこれを簡単にします。
ステップ1：別のサーバーを作成する（weather_server.py）
SSEトランスポートを使用して実行される天気サーバーを作成します。
# weather_server.py
from mcp.server.fastmcp import FastMCP
import uvicorn # 必要：pip install uvicorn
mcp = FastMCP("Weather")
@mcp.tool()
async def get_weather(location: str) -> str:
"""場所の天気を取得します。"""
print(f"Executing get_weather({location})")
# 実際のシナリオでは、天気APIを呼び出します
return f"{location}では常に晴れです"
if __name__ == "__main__":
# SSEトランスポートを使用してサーバーを実行します（uvicornのようなASGIサーバーが必要）
# mcpライブラリは暗黙的にSSE用のFastAPIアプリを作成します。
# デフォルトでは、ポート8000で/sseエンドポイントで実行されます。
print("ポート8000でSSE経由でWeather MCPサーバーを起動しています...")
# uvicorn.run(mcp.app, host="0.0.0.0", port=8000) # 手動で実行できます
mcp.run(transport="sse", host="0.0.0.0", port=8000) # またはmcp.runの便利さを使用
これをweather_server.pyとして保存してください。
ステップ2：クライアントを更新してMultiServerMCPClientを使用する（multi_client_app.py）
import asyncio
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
# --- 重要：パスを更新 ---
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# 天気サーバーは別に実行され、URL経由で接続します
# ---
async def main():
model = ChatOpenAI(model="gpt-4o")
# 複数のサーバーへの接続を定義
server_connections = {
"math_service": { # この接続のユニーク名
"transport": "stdio",
"command": "python",
"args": [math_server_script_path],
# 他のStdioConnectionパラメータも必要に応じて追加
},
"weather_service": { # この接続のユニーク名
"transport": "sse",
"url": "http://localhost:8000/sse", # weather_serverが実行されているURL
# 他のSSEConnectionパラメータも必要に応じて追加
}
}
print("複数のMCPサーバーに接続しています...")
# MultiServerMCPClientコンテキストマネージャを使用
async with MultiServerMCPClient(server_connections) as client:
print("接続が確立されました。")
# 接続されたすべてのサーバーからすべてのツールを取得
all_tools = client.get_tools()
print(f"読み込まれたツール: {[tool.name for tool in all_tools]}")
# 結合されたツールリストを使用してエージェントを作成
agent = create_react_agent(model, all_tools)
# --- エージェントと対話 ---
print("\n数学のクエリのためにエージェントを呼び出しています...")
math_inputs = {"messages": [("human", "what's (3 + 5) * 12?")]}
math_response = await agent.ainvoke(math_inputs)
print("数学の応答:", math_response['messages'][-1].content)
print("\n天気のクエリのためにエージェントを呼び出しています...")
weather_inputs = {"messages": [("human", "nycの天気はどうですか？")]}
weather_response = await agent.ainvoke(weather_inputs)
print("天気の応答:", weather_response['messages'][-1].content)
# --- 例：プロンプトを取得 ---
# print("\n数学サーバープロンプトを取得しています...")
# prompt_messages = await client.get_prompt(
# server_name="math_service", # 接続で定義された名前を使用
# prompt_name="configure_assistant",
# arguments={"skills": "basic arithmetic"}
# )
# print("プロンプト:", prompt_messages)
if __name__ == "__main__":
# 別のターミナルでまず天気サーバーを起動します：
# python weather_server.py
# その後、このクライアントスクリプトを実行します：
asyncio.run(main())
これをmulti_client_app.pyとして保存してください。
実行するには：
- 1つのターミナルで天気サーバーを起動します：
python weather_server.py - 別のターミナルでマルチクライアントアプリを実行します：
python multi_client_app.py
MultiServerMCPClientはmath_server.pyサブプロセスを起動（stdio）し、実行中のweather_server.py（sse）に接続します。ツール（add、multiply、get_weather）を集約し、それらはLangGraphエージェントに利用可能になります。
LangGraph APIサーバーとの統合
MCPツールを使用してLangGraphエージェントを持続的なAPIサービスとしてデプロイできます。鍵は、MultiServerMCPClientのライフサイクルをLangGraphアプリケーションコンテキスト内で正しく管理することです。
次のようなgraph.pyファイルを作成します：
# graph.py
from contextlib import asynccontextmanager
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI # またはAnthropicなど
# --- 重要：パスを更新 ---
# LangGraphサーバーが実行される場所に対するサーバーの相対パスを仮定
math_server_script_path = os.path.abspath("math_server.py")
# ---
# 接続を定義（サーバー環境のためにパス/URLが正しいことを確認）
server_connections = {
"math_service": {
"transport": "stdio",
"command": "python",
"args": [math_server_script_path],
},
"weather_service": {
"transport": "sse",
"url": "http://localhost:8000/sse", # 天気サーバーは独立して実行されている必要があります
}
}
model = ChatOpenAI(model="gpt-4o")
# クライアントのセットアップ/ティアダウンを処理するために非同期コンテキストマネージャを使用
@asynccontextmanager
async def lifespan(_app): # LangGraphはこの構造を期待しています
async with MultiServerMCPClient(server_connections) as client:
print("ライフスパン内でMCPクライアントを初期化しました。")
# クライアントがアクティブなコンテキスト内でエージェントを作成
agent = create_react_agent(model, client.get_tools())
yield {"agent": agent} # エージェントを利用可能にする
# ライフスパンがそれを提供する場合は、別のメイングラフ定義は不要
次のようにあなたのlanggraph.json（またはpyproject.tomlの下の[tool.langgraph]）を設定して、このグラフ定義をライフスパンマネージャと共に使用します：
// langgraph.json（例）
{
"dependencies": ["."], // または必要なパッケージを指定
"graphs": {
"my_mcp_agent": {
"entrypoint": "graph:agent", // ライフスパンから提供されるキーを参照
"lifespan": "graph:lifespan"
}
}
}
これで、langgraph upを実行すると、lifespan関数が実行され、MultiServerMCPClient（およびstdio数学サーバー）が起動します。このコンテキスト内で作成されたエージェントがLangGraphによって提供されます。SSE天気サーバーは別に実行する必要があることに注意してください。
サーバートランスポート（stdio vs. SSE）
stdio：
- 通信：サーバープロセスの標準入力および出力ストリームを介して行われます。
- 利点：ローカル開発のためのシンプルなセットアップ。クライアントがサーバーライフサイクルを管理します。ネットワーキングは関与しません。
- 欠点：密に結合されているため、分散システムや非Pythonサーバーにはあまり適していません。
commandとargsの設定が必要です。
sse（サーバー送信イベント）：
- 通信：HTTPを介してSSEプロトコルを使用します。サーバーはWebサービスとして実行されます（通常はFastAPI/Uvicornを使用しています）。
- 利点：標準的なWebプロトコルであり、ネットワークのサーバーや異なる言語で実装されたサーバーに適しています。サーバーは独立して実行されます。
- 欠点：サーバーは別に実行される必要があります。
urlの設定が必要です。
デプロイニーズに基づいてトランスポートを選択してください。
langchain-mcp-adaptersセットアップのための高度なクライアント設定
MultiServerMCPClient内のStdioConnectionとSSEConnection辞書は、より細かく制御するために追加のオプションのパラメータを受け入れます：
- Stdio：
env（サブプロセス用のカスタム環境変数）、cwd（作業ディレクトリ）、encoding、encoding_error_handler、session_kwargs（mcp.ClientSessionに渡されます）。 - SSE：
headers（カスタムHTTPヘッダー）、timeout（HTTP接続タイムアウト）、sse_read_timeout、session_kwargs。
詳細については、langchain_mcp_adapters/client.py内のMultiServerMCPClientの定義を参照してください。
結論（100語）
langchain-mcp-adaptersライブラリは、標準化されたモデルコンテキストプロトコルと柔軟なLangChainエコシステムとのギャップを効果的に埋めます。MultiServerMCPClientと自動ツール変換を提供することで、開発者は多様なMCP準拠ツールをLangChainエージェントやLangGraphアプリケーションに簡単に組み込むことができます。
基本的なワークフローは次のようになります：
@mcp.tool()を使用してMCPサーバー内でツール（およびオプションでプロンプト）を定義します。- 各サーバーの接続詳細（stdioまたはsse）を含む
MultiServerMCPClientを構成します。 client.get_tools()を介して接続し、ツールを取得するためにクライアントコンテキストマネージャ（async with ...）を使用します。- 取得したLangChain互換ツールをエージェントに渡します（
create_react_agentまたはカスタムエージェント）。
これにより、標準化されたプロトコルを介して専門的な外部ツールを活用する強力でモジュール式のAIアプリケーションを構築できます。さらなる洞察のために、リポジトリ内の例やテストを探求してください。