wrap_tool_call() — langchain Function Reference

Architecture documentation for the wrap_tool_call() function in tool_retry.py from the langchain codebase.

Function python LangChainCore ApiManagement calls 2

Entity Profile

LangChainCore→ ApiManagement→ wrap_tool_call() — langchain Function Reference

Dependency Diagram

graph TD
  731777b2_3052_874d_7b48_c5d58d2e0a8c["wrap_tool_call()"]
  33f83a05_ad1b_0c1c_e484_ac91455db157["ToolRetryMiddleware"]
  731777b2_3052_874d_7b48_c5d58d2e0a8c -->|defined in| 33f83a05_ad1b_0c1c_e484_ac91455db157
  5665de07_b809_7bb1_4e7f_e0409866de63["_should_retry_tool()"]
  731777b2_3052_874d_7b48_c5d58d2e0a8c -->|calls| 5665de07_b809_7bb1_4e7f_e0409866de63
  7a031c34_097e_d25a_0210_cf5d0d368b98["_handle_failure()"]
  731777b2_3052_874d_7b48_c5d58d2e0a8c -->|calls| 7a031c34_097e_d25a_0210_cf5d0d368b98
  style 731777b2_3052_874d_7b48_c5d58d2e0a8c fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/tool_retry.py lines 288–344

    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
    ) -> ToolMessage | Command[Any]:
        """Intercept tool execution and retry on failure.

        Args:
            request: Tool call request with call dict, `BaseTool`, state, and runtime.
            handler: Callable to execute the tool (can be called multiple times).

        Returns:
            `ToolMessage` or `Command` (the final result).

        Raises:
            RuntimeError: If the retry loop completes without returning. This should not happen.
        """
        tool_name = request.tool.name if request.tool else request.tool_call["name"]

        # Check if retry should apply to this tool
        if not self._should_retry_tool(tool_name):
            return handler(request)

        tool_call_id = request.tool_call["id"]

        # Initial attempt + retries
        for attempt in range(self.max_retries + 1):
            try:
                return handler(request)
            except Exception as exc:
                attempts_made = attempt + 1  # attempt is 0-indexed

                # Check if we should retry this exception
                if not should_retry_exception(exc, self.retry_on):
                    # Exception is not retryable, handle failure immediately
                    return self._handle_failure(tool_name, tool_call_id, exc, attempts_made)

                # Check if we have more retries left
                if attempt < self.max_retries:
                    # Calculate and apply backoff delay
                    delay = calculate_delay(
                        attempt,
                        backoff_factor=self.backoff_factor,
                        initial_delay=self.initial_delay,
                        max_delay=self.max_delay,
                        jitter=self.jitter,
                    )
                    if delay > 0:
                        time.sleep(delay)
                    # Continue to next retry
                else:
                    # No more retries, handle failure
                    return self._handle_failure(tool_name, tool_call_id, exc, attempts_made)

        # Unreachable: loop always returns via handler success or _handle_failure
        msg = "Unexpected: retry loop completed without returning"
        raise RuntimeError(msg)