Implementation:Langchain ai Langgraph ToolNode Init
| Attribute | Value |
|---|---|
| API | ToolNode.__init__
|
| Workflow | ReAct_Agent_Creation |
| Type | API Doc |
| Repository | Langchain_ai_Langgraph |
| Source File | libs/prebuilt/langgraph/prebuilt/tool_node.py
|
| Source Lines | L737-781 |
Overview
ToolNode is the execution engine for tools within LangGraph workflows. Its __init__ method accepts a sequence of tools (as BaseTool instances or plain callables) and configuration for error handling, naming, and execution wrapping. During initialization, plain callables are automatically converted to BaseTool instances, and injection metadata is precomputed for each tool so that InjectedState, InjectedStore, and ToolRuntime parameters are resolved efficiently at execution time.
This page also covers the InjectedState, InjectedStore, and ToolRuntime annotation classes that control how runtime context is supplied to tools.
Description
The ToolNode.__init__ method performs the following steps:
- Calls the parent
RunnableCallableconstructor with sync (_func) and async (_afunc) entry points. - Iterates over the provided tools, converting any plain callables to
BaseToolviacreate_tool. - Builds an internal
_tools_by_namedictionary mapping tool names to theirBaseToolinstances. - Precomputes an
_injected_argsmapping for each tool, identifying parameters annotated withInjectedState,InjectedStore, orToolRuntime. - Stores error handling, messages key, and optional wrapper configurations.
The InjectedState class is an annotation that marks tool parameters for automatic state injection. It accepts an optional field argument; when None, the entire state dict is injected, and when a string is provided, only that specific field value is injected.
The InjectedStore class marks parameters for persistent store injection, providing tools with access to the BaseStore instance configured at graph compilation.
The ToolRuntime dataclass bundles state, context, config, stream_writer, tool_call_id, and store into a single object, automatically injected when a tool declares a parameter typed as ToolRuntime.
Usage
from langgraph.prebuilt import ToolNode, InjectedState, InjectedStore
from langgraph.prebuilt.tool_node import ToolRuntime
from langchain_core.tools import tool
from typing import Annotated, Any
@tool
def calculator(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@tool
def state_aware(x: int, state: Annotated[dict, InjectedState]) -> str:
"""Tool that reads graph state."""
return f"Messages: {len(state['messages'])}, x={x}"
@tool
def field_aware(x: int, foo: Annotated[str, InjectedState("foo")]) -> str:
"""Tool that reads a specific state field."""
return foo + str(x)
@tool
def store_tool(key: str, store: Annotated[Any, InjectedStore()]) -> str:
"""Tool with persistent storage access."""
result = store.get(("data",), key)
return result.value if result else "Not found"
@tool
def runtime_tool(x: int, runtime: ToolRuntime) -> str:
"""Tool that uses the full runtime context."""
return f"Call {runtime.tool_call_id}: {x}"
# Basic initialization
tool_node = ToolNode([calculator])
# With error handling and injection
tool_node = ToolNode(
[calculator, state_aware, field_aware, store_tool, runtime_tool],
name="tools",
handle_tool_errors=True,
messages_key="messages",
)
Code Reference
Source Location
| File | libs/prebuilt/langgraph/prebuilt/tool_node.py
|
| Class | ToolNode (lines 616-733 for class body, 737-781 for __init__)
|
| InjectedState | Lines 1597-1670 |
| InjectedStore | Lines 1673-1746 |
| ToolRuntime | Lines 1530-1594 |
Signature
class ToolNode(RunnableCallable):
def __init__(
self,
tools: Sequence[BaseTool | Callable],
*,
name: str = "tools",
tags: list[str] | None = None,
handle_tool_errors: bool
| str
| Callable[..., str]
| type[Exception]
| tuple[type[Exception], ...] = _default_handle_tool_errors,
messages_key: str = "messages",
wrap_tool_call: ToolCallWrapper | None = None,
awrap_tool_call: AsyncToolCallWrapper | None = None,
) -> None
class InjectedState(InjectedToolArg):
def __init__(self, field: str | None = None) -> None
class InjectedStore(InjectedToolArg):
pass # No __init__ parameters beyond base class
@dataclass
class ToolRuntime(Generic[ContextT, StateT]):
state: StateT
context: ContextT
config: RunnableConfig
stream_writer: StreamWriter
tool_call_id: str | None
store: BaseStore | None
Import
from langgraph.prebuilt import ToolNode, InjectedState, InjectedStore
from langgraph.prebuilt.tool_node import ToolRuntime
I/O Contract
Input
| Parameter | Type | Default | Description |
|---|---|---|---|
tools |
Callable] | (required) | Sequence of tools to register. Plain callables are auto-converted to BaseTool.
|
name |
str |
"tools" |
Node name for graph identification and debugging. |
tags |
None | None |
Optional metadata tags for filtering and organization. |
handle_tool_errors |
str | Callable[..., str] | type[Exception] | tuple[type[Exception], ...] | Default callable | Error handling strategy. True catches all errors; str returns custom message; callable receives exception and returns string; exception type(s) catch only those types; False disables handling.
|
messages_key |
str |
"messages" |
Key in state dict containing the message list. |
wrap_tool_call |
None | None |
Sync wrapper to intercept tool execution (receives ToolCallRequest and execute callable).
|
awrap_tool_call |
None | None |
Async wrapper for tool execution interception. Falls back to wrap_tool_call if not provided.
|
Output (at execution time)
When invoked as a graph node, ToolNode produces:
- Dict input:
{"messages": [ToolMessage(...), ...]} - List input:
[ToolMessage(...), ...] - Command tools:
[Command(...)]or mixed list
Internal State After Init
| Attribute | Type | Description |
|---|---|---|
_tools_by_name |
dict[str, BaseTool] |
Mapping from tool name to BaseTool instance. |
_injected_args |
dict[str, _InjectedArgs] |
Per-tool mapping of injection requirements (state fields, store, runtime). |
_handle_tool_errors |
varies | Stored error handling configuration. |
_messages_key |
str |
Key for message list in state. |
_wrap_tool_call |
None | Sync wrapper. |
_awrap_tool_call |
None | Async wrapper. |
Usage Examples
Basic Tool Node
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
node = ToolNode([multiply])
# Invoke with direct tool calls
result = node.invoke(
[{"name": "multiply", "args": {"a": 3, "b": 4}, "id": "1", "type": "tool_call"}]
)
# Returns: [ToolMessage(content="12", tool_call_id="1")]
State Injection
from typing import Annotated
from langchain_core.tools import tool
from langchain_core.messages import AIMessage
from langgraph.prebuilt import ToolNode, InjectedState
@tool
def context_tool(query: str, state: Annotated[dict, InjectedState]) -> str:
"""Tool that uses graph state."""
return f"Query: {query}, Messages: {len(state['messages'])}"
node = ToolNode([context_tool])
tool_call = {"name": "context_tool", "args": {"query": "test"}, "id": "1", "type": "tool_call"}
state = {"messages": [AIMessage("", tool_calls=[tool_call])]}
result = node.invoke(state)
# The 'state' parameter is injected automatically; model only sees 'query'
Custom Error Handling
def handle_value_errors(e: ValueError) -> str:
return f"Invalid input: {e}"
node = ToolNode([my_tool], handle_tool_errors=handle_value_errors)