YouTip

@wrap_tool_call gives you similar control capability at the tool execution level as @wrap_model_call — retry, caching, parameter rewriting, and result post-processing. * * * ## Basic Structure The structure of @wrap_tool_call is similar to @wrap_model_call, accepting 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 tool name and parameters # request.tool: the tool object itself # request.state: current state of the Agent # request.runtime: runtime context # Calling handler(request) will actually execute the tool result = handler(request) # result is ToolMessage or Command return result * * * ## Scenario 1: Tool Call Retry Tool execution may fail due to unstable external services, and automatic retry 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 tool call fails""" max_retries =3 last_result =None for attempt in range(max_retries): try: result = handler(request) # Check if it's an error result if hasattr(result,'status')and result.status=="error": if attempt < max_retries - 1: print(f" Tool returned error, retry attempt {attempt + 1}...") continue if attempt >0: print(f" Attempt {attempt + 1}") return result except Exception as e: if attempt < max_retries - 1: import time time.sleep((attempt + 1) * 2) print(f" Exception {e}, retry attempt {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: """Fetch course data fromTutorial TUTORIAL. Args: course_id: Course ID """ global call_count call_count +=1 # Simulate first two failures, third success if call_count <3: raise Exception(f"Network error: Cannot 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=, middleware=, system_prompt="You are the course assistant ofTutorial TUTORIAL.", ) result = agent.invoke({ "messages": [HumanMessage(content="Help me check the information for course python-001")] }) print(f"\\ Final response: {result['messages'].content}") Running result: Exception Network error: Cannot connect to course service (attempt 1), retry attempt 1... Exception Network error: Cannot connect to course service (attempt 2), retry attempt 2... Attempt 3 Final response: Course python-001 is Python3 Basic Tutorial, with 30 chapters, provided for free. * * * ## Scenario 2: Modifying Tool Parameters Dynamically modify parameters before tool execution to achieve 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 (full-width to half-width, remove extra spaces, etc.)""" tool_call = request.tool_call # Only process tool calls containing city parameter if"city"in tool_call.get("args",{}): city = tool_call # Normalize city name: remove spaces, unify case normalized = city.strip().replace(" ","")# Remove full-width spaces # Replace parameter new_args ={**tool_call,"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), you can cache the results: ## 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_name}") cached_content = tool_cache return ToolMessage( content=cached_content, tool_call_id=request.tool_call.get("id",""), name=tool_name, ) # Execute tool result = handler(request) # Store in cache if hasattr(result,'content'): tool_cache= result.content print(f" {tool_name}, current {len(tool_cache)} items") 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 of 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 # Log successful call print(f" {tool_name}({tool_args}) succeeded, took {elapsed:.2f}s") return result except Exception as e: elapsed =time.time() - start_time # Log failed call print(f" {tool_name}({tool_args}) failed, took {elapsed:.2f}s, error: {e}") raise * * * ## Scenario 5: Decide Subsequent Flow Based on Result 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 tool returns empty result, directly end Agent to save model calls""" result = handler(request) # Check if empty result was returned if hasattr(result,'content')and( "Not found"in str(result.content) or"No result"in str(result.content) or"No / None / Not have"in str(result.content) ): # Directly return a Command to update state # Add an AI message explaining the situation from langchain.messages import AIMessage return Command(update={ "messages": [ AIMessage(content="Sorry, no related information found. Please try a different keyword.") ] }) return result > When wrap_tool_call returns a Command, you can modify the Agent state through the update parameter. Using Command allows you to directly append AI messages to the message list, and then the Agent loop will naturally end. * * * ## @wrap_model_call vs @wrap_model_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, degradation, caching, prompt modification | Tool retry, caching, parameter rewriting, result processing |