Custom tool execution: bypassing ToolNode
Why this matters
ToolNode works well for simple cases, but in production you often need conditional tool invocation, custom retry logic, state-aware tool selection, or tool output transformation before storing state. Knowing how to wire tools manually prevents you from fighting the framework when standard patterns don't fit.
Explanation
ToolNode is a convenience wrapper: it takes tool calls from an LLM, invokes the tools, and stores results back in state. When you bypass it, you explicitly implement this flow in a custom node function, giving you control over each step.
Mechanically: instead of adding a ToolNode and connecting it, you write a regular node function that reads tool calls from state (usually state.messages), invokes the tools manually, and mutates state with results. You decide what counts as a tool call, how to handle failures, and how to structure the result before storing it.
Use this when you need conditional tool execution ("only call this tool if a flag is set"), custom retry logic ("call this tool up to 3 times"), tool chaining ("use the output of tool A as input to tool B"), or result validation ("reject this tool call if the output doesn't match expected schema").
Analogy
ToolNode is like a vending machine that automatically dispenses the snack you request. Custom tool execution is reaching behind the machine, selecting which snack to dispense, deciding whether to dispense it based on inventory, and choosing how to package the result before returning it.
Code
import json
from typing import Any
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing_extensions import TypedDict
class State(TypedDict):
messages: list[dict[str, str]]
tool_results: list[dict[str, Any]]
def calculator_tool(operation: str, a: float, b: float) -> float:
"""Simple calculator tool."""
if operation == "add":
return a + b
elif operation == "subtract":
return a - b
elif operation == "multiply":
return a * b
else:
raise ValueError(f"Unknown operation: {operation}")
def llm_node(state: State) -> Command[State]:
"""LLM that requests tool calls."""
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools_schema = [
{
"type": "function",
"function": {
"name": "calculator",
"description": "Performs arithmetic operations",
"parameters": {
"type": "object",
"properties": {
"operation": {"type": "string", "enum": ["add", "subtract", "multiply"]},
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["operation", "a", "b"]
}
}
}
]
response = model.bind(tools=tools_schema).invoke(state["messages"])
return Command(
update={"messages": state["messages"] + [{"role": "assistant", "content": response.content, "tool_calls": response.tool_calls if hasattr(response, 'tool_calls') else []}]},
goto="custom_tool_executor" if (hasattr(response, 'tool_calls') and response.tool_calls) else "end_node"
)
def custom_tool_executor(state: State) -> Command[State]:
"""Custom node that executes tools with conditional logic and error handling."""
messages = state["messages"]
last_message = messages[-1]
tool_calls = last_message.get("tool_calls", [])
results = []
for tool_call in tool_calls:
tool_name = tool_call.get("function", {}).get("name") if isinstance(tool_call, dict) else tool_call.name
args = tool_call.get("function", {}).get("arguments") if isinstance(tool_call, dict) else tool_call.args
try:
if isinstance(args, str):
args = json.loads(args)
if tool_name == "calculator":
operation = args.get("operation")
a = args.get("a")
b = args.get("b")
if operation == "multiply" and (a > 100 or b > 100):
result = {"error": "Multiplying numbers > 100 not allowed", "rejected": True}
else:
output = calculator_tool(operation, a, b)
result = {"result": output, "rejected": False}
else:
result = {"error": f"Unknown tool: {tool_name}", "rejected": True}
except Exception as e:
result = {"error": str(e), "rejected": True}
results.append({"tool_name": tool_name, "tool_input": args, **result})
return Command(
update={
"messages": messages + [{"role": "user", "content": json.dumps(results)}],
"tool_results": state["tool_results"] + results
},
goto="llm_node"
)
def end_node(state: State) -> None:
"""Terminal node."""
pass
graph = StateGraph(State)
graph.add_node("llm_node", llm_node)
graph.add_node("custom_tool_executor", custom_tool_executor)
graph.add_node("end_node", end_node)
graph.add_edge(START, "llm_node")
graph.add_edge("end_node", END)
compiled_graph = graph.compile()
input_state = {
"messages": [{"role": "user", "content": "What is 5 plus 3?"}],
"tool_results": []
}
result = compiled_graph.invoke(input_state)
print(f"Messages: {result['messages']}")
print(f"Tool Results: {result['tool_results']}")Messages: [{'role': 'user', 'content': 'What is 5 plus 3?'}, {'role': 'assistant', 'content': '', 'tool_calls': [{'id': 'call_xyz', 'function': {'name': 'calculator', 'arguments': '{"operation": "add", "a": 5, "b": 3}'}, 'type': 'function'}]}, {'role': 'user', 'content': '[{"tool_name": "calculator", "tool_input": {"operation": "add", "a": 5, "b": 3}, "result": 8.0, "rejected": false}]'}, {'role': 'assistant', 'content': 'The sum of 5 and 3 is 8.'}]
Tool Results: [{'tool_name': 'calculator', 'tool_input': {'operation': 'add', 'a': 5, 'b': 3}, 'result': 8.0, 'rejected': false}] What just happened?
The LLM was called and requested a calculator tool. The custom_tool_executor node intercepted that request, parsed the tool call arguments, applied validation logic (rejecting multiplications over 100), invoked the calculator_tool function directly, and stored the structured result in both the message history and tool_results state. Because this was a simple operation that passed validation, it succeeded and returned a result. The graph then re-invoked the LLM with the tool result so it could generate a final response.
Common gotcha
When you parse tool_calls from LLM responses, the structure varies by provider and model version: sometimes it's response.tool_calls, sometimes the arguments are pre-parsed dicts, sometimes they're JSON strings. Always check the type and parse defensively. Also, don't forget to update state with both the tool result AND information about what was rejected/failed: the LLM needs to see failures to decide whether to retry or give up.
Error recovery
json.JSONDecodeErrorKeyError on tool_call fieldsTool never gets called again after rejectionExperienced dev note
In production, the temptation is to silence tool errors or store them separately. Resist this: always feed failures back into the message thread so the LLM knows what went wrong. This is how agentic systems recover from transient failures and decide whether to retry. Also, add a max tool call depth counter to state to prevent infinite tool loops; an LLM can keep requesting the same failed tool over and over. Set a limit like max_iterations=10 and fail gracefully.
Check your understanding
If an LLM makes a tool call that your validation logic rejects, how does the LLM know it was rejected so it can try a different approach? What state change must happen for the LLM to see the rejection?
Show answer hint
A correct answer explains that the rejection message must be appended to the messages list in state, and the graph must route back to the LLM node (not to END) so the LLM can read the rejection and adjust its next tool call.