Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Langchain ai Langgraph ToolNode Init

From Leeroopedia
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:

  1. Calls the parent RunnableCallable constructor with sync (_func) and async (_afunc) entry points.
  2. Iterates over the provided tools, converting any plain callables to BaseTool via create_tool.
  3. Builds an internal _tools_by_name dictionary mapping tool names to their BaseTool instances.
  4. Precomputes an _injected_args mapping for each tool, identifying parameters annotated with InjectedState, InjectedStore, or ToolRuntime.
  5. 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)

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment