LangChain Tool Call Interception -- @wrap_tool_call
@wrap_tool_call allows you to implement control capabilities similar to @wrap_model_call at the tool execution level—retry, caching, parameter rewriting, and result post-processing.
Basic Structure
The structure of @wrap_tool_call is similar to @wrap_model_call, receiving two parameters: request and handler:
Example
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: Contains the tool name and parameters
# request.tool: The tool object itself
# request.state: The Agent's current state
# request.runtime: Runtime context
# The tool is actually executed only when handler(request) is called
result = handler(request)
# result is a ToolMessage or Command
return result
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: Contains the tool name and parameters
# request.tool: The tool object itself
# request.state: The Agent's current state
# request.runtime: Runtime context
# The tool is actually executed only when handler(request) is called
result = handler(request)
# result is a ToolMessage or Command
return result
Scenario 1: Tool Call Retry
Tool execution may fail due to unstable external services; automatic retries can improve reliability:
Example
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Automatically retry when a tool call fails"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Check if it is an error result
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Retry] Tool returned an error, retry #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Retry Successful] Attempt #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Retry] Exception {e}, retry #{attempt + 1}...")
else:
raise
return last_result
# Simulate a tool that may fail
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Get course data from EXAMPLE.
Args:
course_id: Course ID
"""
global call_count
call_count += 1
# Simulate failure for the first two attempts, success on the third
if call_count < 3:
raise Exception(f"Network error: unable to connect to course service (attempt #{call_count})")
return f"Course {course_id}: Python3 Basic Tutorial, 30 chapters, free"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="You are the course assistant for EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Help me look up the information for course python-001")]
})
print(f"\nFinal reply: {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Automatically retry when a tool call fails"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Check if it is an error result
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Retry] Tool returned an error, retry #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Retry Successful] Attempt #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Retry] Exception {e}, retry #{attempt + 1}...")
else:
raise
return last_result
# Simulate a tool that may fail
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Get course data from EXAMPLE.
Args:
course_id: Course ID
"""
global call_count
call_count += 1
# Simulate failure for the first two attempts, success on the third
if call_count < 3:
raise Exception(f"Network error: unable to connect to course service (attempt #{call_count})")
return f"Course {course_id}: Python3 Basic Tutorial, 30 chapters, free"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="You are the course assistant for EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Help me look up the information for course python-001")]
})
print(f"\nFinal reply: {result['messages'][-1].content}")
Output:
[重试] 异常 网络错误：无法连接到课程服务（第 1 次尝试），1 次重试... [重试] 异常 网络错误：无法连接到课程服务（第 2 次尝试），2 次重试... [重试成功] 第 3 次尝试 Final reply: Course python-001 is the Python3 basics tutorial, 30 chapters, free of charge.
Scenario 2: Modifying Tool Parameters
Dynamically modifying parameters before tool execution allows parameter transformation without modifying the tool code:
Example
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def normalize_city_name(request, handler):
"""Automatically normalize city names (convert full-width to half-width, remove extra spaces, etc.)"""
tool_call = request.tool_call
# Only process tool calls that contain the city parameter
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Normalize city name: remove spaces, unify case
normalized = city.strip().replace(" ", "") # Remove full-width spaces
# Replace the parameter
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
@wrap_tool_call
def normalize_city_name(request, handler):
"""Automatically normalize city names (convert full-width to half-width, remove extra spaces, etc.)"""
tool_call = request.tool_call
# Only process tool calls that contain the city parameter
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Normalize city name: remove spaces, unify case
normalized = city.strip().replace(" ", "") # Remove full-width spaces
# Replace the parameter
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
Scenario 3: Tool Result Caching
For repeated tool calls (same tool + same parameters), results can be cached:
Example
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Cache tool execution results"""
# Generate cache key: tool name + parameters
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Check cache
if cache_key in tool_cache:
print(f"[Tool cache hit] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Execute the tool
result = handler(request)
# Store in cache
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Tool cache write] {tool_name}, currently {len(tool_cache)} entries")
return result
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Cache tool execution results"""
# Generate cache key: tool name + parameters
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Check cache
if cache_key in tool_cache:
print(f"[Tool cache hit] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Execute the tool
result = handler(request)
# Store in cache
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Tool cache write] {tool_name}, currently {len(tool_cache)} entries")
return result
Scenario 4: Tool Call Logging and Monitoring
Record detailed information for all tool calls:
Example
import time
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Monitor performance metrics for tool calls"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Record start time
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Record successful call
print(f"[Monitor] {tool_name}({tool_args}) succeeded, took {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Record failed call
print(f"[Monitor] {tool_name}({tool_args}) failed, took {elapsed:.2f}s, error: {e}")
raise
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Monitor performance metrics for tool calls"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Record start time
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Record successful call
print(f"[Monitor] {tool_name}({tool_args}) succeeded, took {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Record failed call
print(f"[Monitor] {tool_name}({tool_args}) failed, took {elapsed:.2f}s, error: {e}")
raise
Scenario 5: Deciding Subsequent Flow Based on Results
You can decide whether to continue the Agent loop based on the tool execution result:
Example
from langchain.agents.middleware import wrap_tool_call
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""If the tool returns an empty result, end the Agent directly without wasting model calls"""
result = handler(request)
# Check if an empty result was returned
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Directly return a Command to update state
# Add an AI message to explain the situation
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Sorry, no relevant information was found. Please try a different keyword.")
]
})
return result
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""If the tool returns an empty result, end the Agent directly without wasting model calls"""
result = handler(request)
# Check if an empty result was returned
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Directly return a Command to update state
# Add an AI message to explain the situation
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Sorry, no relevant information was found. Please try a different keyword.")
]
})
return result
When wrap_tool_call returns a Command, you can modify the Agent state via the update parameter. Using a Command allows you to directly append an AI message to the message list, and then the Agent loop will naturally end.
@wrap_model_call vs @wrap_tool_call
| Dimension | @wrap_model_call | @wrap_tool_call |
|---|---|---|
| Interception Target | Model call | Tool execution |
| request content | model、messages、tools、system_prompt | tool_call、tool、state、runtime |
| Return type | ModelResponse or AIMessage | ToolMessage or Command |
| Applicable scenarios | Model retry, fallback, caching, prompt modification | Tool retry, caching, parameter rewriting, result processing |