この記事は人間の血と汗と涙により生成されています。
こんにちは。もうすっかりアドカレの季節ですね。
今年は弊社アドカレのトップバッターを担当します。
トップバッターなのにアドカレで何書くかはまだ決めてなくて正直焦っています。
焦るくらいならこの記事を充てたらいいんじゃないかと思ったのですが、それはなんか違う気がしたのでアドカレ前に本記事を投稿しちゃうという思い切ったことをしています。
それでは本題。
やりたいこと
タイトルの通り LangGraph で定義している ToolNode で MCP から提供されるツールを呼び出したい、というのが今回のやりたいことです。
ToolNode は langgraph-prebuilt パッケージが提供する LangGraph の組み込みコンポーネントです。AIメッセージに含まれるツール呼び出しを自動的に実行し、結果を ToolMessage として返すノードです。
そこで MCP 経由でツールを呼び出したいというのが今回のやりたいことです。
そもそも LangGraph ってなんのこっちゃ、という方のために少しだけ前提からお話ししたいと思います。
動作確認環境
| 項目 | バージョン |
|---|---|
| WSL | Ubuntu-24.04 |
| Python | 3.12.11 |
| langchain-aws | 1.0.0 |
| langchain-core | 1.0.5 |
| langchain-mcp-adapters | 0.1.13 |
| langgraph | 1.0.3 |
LangGraph とは
LangGraph とは、LLM を中心としたワークフローをグラフ構造で構築・管理するためのフレームワークです。
グラフ構造で構築・管理するというのが大きな特徴です。ワークフローに何かをインプットしたら何かがアウトプットされる感じです。
最も簡単なグラフの例を見てみましょう。
+-----------+
| __start__ |
+-----------+
*
*
*
+-------+
| agent |
+-------+
*
*
*
+---------+
| __end__ |
+---------+
これが Graph の一例です。ワークフローなので必ず start と end があります。
このワークフローは start の後、処理が agent（という名前がついたノード） に託され、agent での処理が終わったら end になります。なので実際はフローにはなってないグラフです。インプットしたものが agent ノードで処理されて終わりです。
各ポイントが Node（ノード）と呼ばれ、それらを結んでいる線が Edge（エッジ）と呼ばれます。
上記グラフの具体的なコードはこんな感じです。
コード全量
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.graph.state import CompiledStateGraph
from utils import print_messages
llm = ChatBedrockConverse(
model="global.anthropic.claude-sonnet-4-5-20250929-v1:0",
region_name="ap-northeast-1",
)
def call_model(state: MessagesState) -> dict:
messages = state["messages"]
response = llm.invoke(messages)
return {"messages": [response]}
def create_graph() -> CompiledStateGraph:
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)
return workflow.compile()
def main() -> None:
graph = create_graph()
result = graph.invoke(
{"messages": [HumanMessage(content="1.5と3の加算を行ってください。")]}
)
print_messages(result)
if __name__ == "__main__":
main()
実行してみます。
実行結果
$ python main_simple.py
[1] Human:
1.5と3の加算を行ってください。
[2] AI:
Content: 1.5 + 3 = **4.5**
※出力を見やすくするために utils.print_messages を個別に定義しています。念のためこちらのソースも載せておきます。
utils.py
def print_messages(result: dict) -> None:
"""メッセージを人間にとってわかりやすく出力する"""
messages = result.get("messages", [])
for i, msg in enumerate(messages, 1):
msg_type = type(msg).__name__
if msg_type == "HumanMessage":
print(f"\n[{i}] Human:")
print(f" {msg.content}")
elif msg_type == "AIMessage":
print(f"\n[{i}] AI:")
if msg.content:
print(f" Content: {msg.content}")
if hasattr(msg, "tool_calls") and msg.tool_calls:
print(" Tool Calls:")
for tool_call in msg.tool_calls:
print(
f" - {tool_call.get('name', 'unknown')}({tool_call.get('args', {})})"
)
elif msg_type == "ToolMessage":
print(f"\n[{i}] Tool:")
print(f" Name: {msg.name if hasattr(msg, 'name') else 'unknown'}")
print(f" Result: {msg.content}")
else:
print(f"\n[{i}] ❓ {msg_type}:")
print(f" {msg}")
こんな感じでインプット「1.5と3の加算を行ってください。」に対して最終的にはアウトプット「1.5 + 3 = 4.5」が得られました。
ちなみに LangGraph はつい最近 2025年10月22日 に正式リリースされているほやほやのフレームワークです。といっても、正式リリース前からたくさんのユーザーに支持されています。私の上記説明だけではかなり不足しているので個人的にわかりやすいと思った記事を引用します。
ツールを呼び出す
上記の例では「1.5と3の加算を行ってください。」というリクエストに対して、LLM（= agent ノード）が自分自身で「1.5 + 3 = 4.5」という計算をして回答を導きました。
今回使用しているモデルは global.anthropic.claude-sonnet-4-5-20250929-v1:0 という現時点ではかなり新しいモデル（厳密には AWS の Bedrock 経由で Anthropic 社の Claude Sonnet 4.5 というモデルを呼び出しています）になるのでそれくらいは朝飯前なのですが、毎回確実に計算してほしいのでそういう時は LLM 自身による推論ではなく、ツールを与えてそれを実行して計算してもらいます。
グラフはこんな感じになります。
+-----------+
| __start__ |
+-----------+
*
*
*
+-------+
| agent |
+-------+.
. .
.. ..
. .
+---------+ +-------+
| __end__ | | tools |
+---------+ +-------+
agent ノードから tools というノードに線が追加されました。
上記グラフの具体的なコードはこんな感じです。
コード全量
import operator
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from utils import print_messages
# 計算ツールを定義
@tool
def calculate(operation: str, a: float, b: float) -> str:
"""2つの数値で計算を実行する
Args:
operation (str): 操作の種類(add, subtract, multiply, divide)
a (float): 1つ目の数値
b (float): 2つ目の数値
Returns:
str: 計算結果またはエラーメッセージ
"""
ops = {
"add": (operator.add, "+"),
"subtract": (operator.sub, "-"),
"multiply": (operator.mul, "*"),
"divide": (operator.truediv, "/"),
}
if operation not in ops:
return f"Error: Unknown operation '{operation}'"
func, symbol = ops[operation]
if operation == "divide" and b == 0:
return "Error: Division by zero"
result = func(a, b)
return f"Result: {a} {symbol} {b} = {result}"
tools = [calculate]
llm = ChatBedrockConverse(
model="global.anthropic.claude-sonnet-4-5-20250929-v1:0",
region_name="ap-northeast-1",
)
llm_with_tools = llm.bind_tools(tools) # LLMの呼び出し時にツールの情報を渡す
def call_model(state: MessagesState) -> dict:
messages = state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
def create_graph() -> CompiledStateGraph:
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools)) # ToolNode を追加
workflow.add_edge(START, "agent")
workflow.add_conditional_edges( # add_conditional_edges で LLM がツールを呼ぶ必要があると判断したら ToolNode に遷移
"agent",
tools_condition,
)
workflow.add_edge("tools", "agent")
workflow.add_edge("agent", END)
return workflow.compile()
def main() -> None:
graph = create_graph()
result = graph.invoke(
{"messages": [HumanMessage(content="1.5と3の加算を行ってください。")]}
)
print_messages(result)
if __name__ == "__main__":
main()
calculate という四則演算ツールを定義して agent が利用できるように紐づけています。
これにより LLM を呼び出したとき LLM が「ツールの実行が必要」と判断したらツールが実行出来る仕組みになっています。
実行してみましょう。
実行結果
$ python main_tools.py
[1] Human:
1.5と3の加算を行ってください。
[2] AI:
Content: [{'type': 'tool_use', 'name': 'calculate', 'input': {'operation': 'add', 'a': 1.5, 'b': 3}, 'id': 'tooluse_JdjwNAuBQDq03Bzf43N9Jw'}]
Tool Calls:
- calculate({'operation': 'add', 'a': 1.5, 'b': 3})
[3] Tool:
Name: calculate
Result: Result: 1.5 + 3.0 = 4.5
[4] AI:
Content: 1.5と3の加算結果は **4.5** です。
ツールを実行して回答を導けていそうです。少しだけ流れを解説すると
[1] 「1.5と3の加算を行ってください。」というリクエストが来ました。
[2] 「calculate というツールを {'operation': 'add', 'a': 1.5, 'b': 3} という引数で実行する必要がある」と LLM が判断します。
[3] calculate ツールが実行され、結果が出力されます。
[4] ツールの結果をもとに LLM が回答します。
ツールがないときとの違いですが、このとき LLM は自分で計算していません。ツールが計算した結果を参照して回答しているだけになります。
これがいわゆる Tool Calling や Function Calling と呼ばれる仕組みになります。
ToolNode で MCP を呼び出したい
ようやく本題です。今回は MCP で提供されるツールをこの ToolNode で呼び出せるようにしたいです。
langchain_mcp_adapters ライブラリを利用することでこれが実現できます。
MCPサーバーを用意する
デモのためにまず MCP サーバーを構築しましょう。先述の例が四則演算ツールの例だったので、同じ四則演算ツールを提供する MCP サーバーを構築します。
こんな感じで FastMCP を使ってローカルで起動させます。
通信方式は streamable-http と stdio の2つが主にありますが、今回は前者で起動させます。
コード全量
import operator
from mcp.server.fastmcp import FastMCP # Create MCP server instance
mcp = FastMCP("Calculate", host="localhost", port="8000", debug=True, log_level="DEBUG")
@mcp.tool()
def calculate(operation: str, a: float, b: float) -> str:
"""2つの数値で計算を実行する
Args:
operation (str): 操作の種類(add, subtract, multiply, divide)
a (float): 1つ目の数値
b (float): 2つ目の数値
Returns:
str: 計算結果またはエラーメッセージ
"""
ops = {
"add": (operator.add, "+"),
"subtract": (operator.sub, "-"),
"multiply": (operator.mul, "*"),
"divide": (operator.truediv, "/"),
}
if operation not in ops:
return f"Error: Unknown operation '{operation}'"
func, symbol = ops[operation]
if operation == "divide" and b == 0:
return "Error: Division by zero"
result = func(a, b)
return f"Result: {a} {symbol} {b} = {result}"
if __name__ == "__main__":
mcp.run(transport="streamable-http")
起動してみます。
コンソールログ
$ python mcp_server.py
INFO: Started server process [31403]
INFO: Waiting for application startup.
StreamableHTTP session manager started
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)
無事ローカルホストに port 8000 で起動できました。
ToolNode で MCP を呼び出す
MCP プロトコル自体が非同期の設計になっているので、ToolNode で MCP を利用したいなら呼び出し側も合わせて非同期にする必要があります。
そのため ToolNode で MCP を呼び出すには次のように処理を個々の非同期化する必要があります。
コード全量
import asyncio
from contextlib import asynccontextmanager
from typing import Any
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage
from langchain_core.tools import BaseTool
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from mcp.types import TextContent
from utils import print_messages
# =============================================================================
# MCPクライアントの設定
# =============================================================================
class MCPClientManager:
"""MCPサーバーとの接続を管理するクラス"""
def __init__(self, base_url: str = "http://localhost:8000/mcp"):
self.base_url = base_url
self.session: ClientSession | None = None
@asynccontextmanager
async def connect(self):
"""MCPサーバーに接続（Streamable HTTP）"""
async with streamablehttp_client(self.base_url) as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
self.session = session
yield session
mcp_manager = MCPClientManager()
# =============================================================================
# LangGraphの設定
# =============================================================================
llm = ChatBedrockConverse(
model="global.anthropic.claude-sonnet-4-5-20250929-v1:0",
region_name="ap-northeast-1",
)
def create_graph(tools: list[BaseTool]) -> CompiledStateGraph:
llm_with_tools = llm.bind_tools(tools)
async def call_model(state: MessagesState) -> dict:
"""非同期でLLMを呼び出す"""
messages = state["messages"]
response = await llm_with_tools.ainvoke(messages) # 非同期呼び出し
return {"messages": [response]}
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools))
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", tools_condition)
workflow.add_edge("tools", "agent")
return workflow.compile()
async def main() -> None:
async with mcp_manager.connect():
tools = await load_mcp_tools(mcp_manager.session) # 非同期呼び出し
graph = create_graph(tools)
result = await graph.ainvoke( # 非同期呼び出し
{"messages": [HumanMessage(content="1.5と3の加算を行ってください。")]}
)
print_messages(result)
if __name__ == "__main__":
asyncio.run(main())
MCP まわりのポイントを解説していきます。
class MCPClientManager:
"""MCPサーバーとの接続を管理するクラス"""
def __init__(self, base_url: str = "http://localhost:8000/mcp"):
self.base_url = base_url
self.session: ClientSession | None = None
ここでは MCP サーバーへの接続を管理するラッパークラスを定義しています。base_url で MCP サーバーのエンドポイントを指定しています。
これがいわゆる MCP クライアントになります。
@asynccontextmanager
async def connect(self):
async with streamablehttp_client(self.base_url) as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
self.session = session
yield session
ここが MCP 接続の核心部分です。
streamablehttp_client でトランスポート層の接続を確立します。 stdio の通信方式の場合はここを変更する必要があります。
次に ClientSession で MCPプロトコル層の接続を確立します。read と write を受け取っていますが、これはお約束みたいなものです。
await session.initialize() でハンドシェイク（クライアント側とサーバー側の挨拶みたいなもの）をしてます。これによって、MCP サーバーで何が出来るのかをクライアント側が把握できるようになります。ここでは把握できるのは利用できるツールそのものではなく、session.list_tools() でツールの一覧が表示できることや session.call_tool(...) でツールを呼び出しできる、といった情報が取得されます。
async def main() -> None:
async with mcp_manager.connect():
tools = await load_mcp_tools(mcp_manager.session)
graph = create_graph(tools)
result = await graph.ainvoke(
{"messages": [HumanMessage(content="1.5と3の加算を行ってください。")]}
)
print_messages(result)
main の中では先ほど定義した MCPClientManager を利用してコネクションを確立し、 tools = await load_mcp_tools(mcp_manager.session) で利用可能なツールの一覧を取得します。
load_mcp_tools は langchain_mcp_adapters が提供している関数で、MCP サーバーから取得したツール定義を LangGraph で利用可能な BaseTool 形式に変換してくれています。
それをもとにグラフを定義すれば、MCP サーバーから取得したツールを ToolNode で呼び出せるというわけです。
実行してみましょう。
実行結果
$ python main_mcp.py
[1] Human:
1.5と3の加算を行ってください。
[2] AI:
Content: [{'type': 'tool_use', 'name': 'calculate', 'input': {'operation': 'add', 'a': 1.5, 'b': 3}, 'id': 'tooluse_JNJ9l-8QT72kWXnAj4gOAw'}]
Tool Calls:
- calculate({'operation': 'add', 'a': 1.5, 'b': 3})
[3] Tool:
Name: calculate
Result: Result: 1.5 + 3.0 = 4.5
[4] AI:
Content: 1.5と3の加算結果は**4.5**です。
自分で定義した tool をバインドしたときと全く同様に、ツールを実行して回答を得ることが出来ました。
MCP サーバー側のコンソールログを見ると ListToolsRequest や CallToolRequest が来ているのがわかります。
MCP サーバー側のコンソールログ
$ python mcp_server.py
INFO: Started server process [31403]
INFO: Waiting for application startup.
StreamableHTTP session manager started
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)
Created new transport with session ID: 5da16f64a42144ff8cb0ccf2b062b633
INFO: 127.0.0.1:48304 - "POST /mcp HTTP/1.1" 200 OK
INFO: 127.0.0.1:48308 - "GET /mcp HTTP/1.1" 200 OK
INFO: 127.0.0.1:48306 - "POST /mcp HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:48310 - "POST /mcp HTTP/1.1" 200 OK
Processing request of type ListToolsRequest
INFO: 127.0.0.1:48320 - "POST /mcp HTTP/1.1" 200 OK
Processing request of type CallToolRequest
Terminating session: 5da16f64a42144ff8cb0ccf2b062b633
無事、LangGraph の ToolNode で MCP を呼び出すことが出来ました。
さいごに
MCP プロトコル自体が非同期の設計になっているので、ToolNode で MCP を利用したいなら呼び出し側も合わせて非同期にする必要があります。
というのが、私はわかっておらず最初つまずきました。他にも私と同じようにつまずいた方の助けになれば幸いです。
ここまで読んでいただきありがとうございました。
さて、アドカレのネタはどうしよう。