Module 2: Tools and Tool Calling

Error Handling and Troubleshooting

Capsule overview

Tools fail. External APIs go down, models invent names for tools that don't exist, arguments arrive with the wrong type, and timeouts hit exactly when you need the answer most. If your tool calling system doesn't handle these failures, a single broken tool can bring down the entire application.

In this capsule you'll learn to build robust tool calling that survives real-world errors. You'll see the five most common error types in tool calling, how to validate arguments before running a tool, how to wrap executions in try/except with retry logic and exponential backoff, and how to report errors in a structured way with ToolException. You'll also learn to debug tool calls by inspecting the messages the model generates — a skill you'll use constantly whenever something doesn't work the way you expect.

The difference between a prototype and a production system isn't functionality: it's how it handles failure. After this capsule, your tools will handle errors the way a professional service would.


The 5 most common errors in tool calling

Before you learn to handle errors, you need to recognize them. These are the five you'll run into most often:

1. Tool not found (the model hallucinates a name)

The model decides to call a tool that doesn't exist. This happens when the model "invents" a tool name based on its general training instead of sticking to the tools you bound with bind_tools.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny, 25°C in {city}"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])

response = model_with_tools.invoke("Search Google for information about Python")
print(response.tool_calls)
# Possible output (the model may try to call a tool that doesn't exist):
# [{'name': 'google_search', 'args': {'query': 'Python'}, 'id': 'call_abc123'}]
# ↑ 'google_search' does NOT exist among our tools

Why it happens: The model knows search tools exist because of its training. Even though you only bound get_weather, it sometimes generates a tool call for a tool it "believes" should be there. This is more frequent with less capable models or when the prompt isn't specific.

2. Invalid arguments (wrong types)

The model passes arguments that don't match the tool's schema. For example, it sends a string where a number is expected, or a single value where a list is expected.

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    operation: str = Field(description="Operation: 'add', 'subtract', 'multiply', 'divide'")
    a: float = Field(description="First number")
    b: float = Field(description="Second number")

@tool(args_schema=CalculatorInput)
def calculator(operation: str, a: float, b: float) -> str:
    """Perform basic math operations."""
    operations = {
        "add": a + b,
        "subtract": a - b,
        "multiply": a * b,
        "divide": a / b if b != 0 else "Error: division by zero",
    }
    if operation not in operations:
        return f"Operation '{operation}' not supported"
    return str(operations[operation])

# The model could pass:
# {'operation': 'sum', 'a': 'ten', 'b': 5}
# 'sum' isn't a valid operation, and 'ten' isn't a float

3. Timeout during execution (a slow external API)

The tool calls an external API that takes too long or never responds. Without a timeout, your application waits forever.

4. Unexpected return format

The tool returns a data type the flow wasn't expecting — for example, None instead of a string, or a dictionary when plain text was expected. The model needs a string to generate its final answer.

5. Rate limiting on external APIs

The external API your tool queries blocks requests for exceeding the rate limit. Unlike the model's rate limiting (which you saw in capsule 07 of Module 1), this one happens in the service your tool calls.


Validate arguments before executing

The first line of defense is verifying that the arguments are valid before running the tool's logic. This avoids unnecessary errors and gives the model clear messages so it can correct itself.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool

VALID_TICKERS = {"AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"}

@tool
def get_stock_price(ticker: str) -> str:
    """Get the current price of a stock. Valid tickers: AAPL, GOOGL, MSFT, AMZN, TSLA."""
    ticker = ticker.upper().strip()

    if not ticker:
        return "Error: empty ticker. Provide a symbol such as 'AAPL'."

    if ticker not in VALID_TICKERS:
        return f"Error: ticker '{ticker}' not recognized. Valid tickers: {', '.join(sorted(VALID_TICKERS))}"

    prices = {"AAPL": 185.50, "GOOGL": 142.30, "MSFT": 420.80, "AMZN": 195.20, "TSLA": 248.60}
    return f"{ticker}: ${prices[ticker]:.2f} USD"

result = get_stock_price.invoke({"ticker": "AAPL"})
print(result)
# Expected output: AAPL: $185.50 USD

result = get_stock_price.invoke({"ticker": "INVALID"})
print(result)
# Expected output: Error: ticker 'INVALID' not recognized. Valid tickers: AAPL, AMZN, GOOGL, MSFT, TSLA

Key point: When you return a descriptive error message (instead of raising an exception), the model receives that message as a ToolMessage and can correct itself — for example, by asking the user to clarify the ticker.


Try/except for tool execution

No matter how much you validate the arguments: execution can fail for external reasons (network, API down, corrupt data). Wrapping execution in try/except is mandatory for any tool that talks to external services.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool
import json

@tool
def fetch_user_data(user_id: str) -> str:
    """Get a user's data by ID."""
    try:
        if not user_id.isdigit():
            return f"Error: user_id must be numeric, I got '{user_id}'"

        users = {
            "1": {"name": "Ana García", "email": "ana@example.com", "role": "admin"},
            "2": {"name": "Carlos López", "email": "carlos@example.com", "role": "user"},
        }

        if user_id not in users:
            return f"Error: user with ID '{user_id}' not found"

        return json.dumps(users[user_id], ensure_ascii=False)

    except Exception as e:
        return f"Unexpected error looking up the user: {type(e).__name__}: {e}"

print(fetch_user_data.invoke({"user_id": "1"}))
# Expected output: {"name": "Ana García", "email": "ana@example.com", "role": "admin"}

print(fetch_user_data.invoke({"user_id": "999"}))
# Expected output: Error: user with ID '999' not found

print(fetch_user_data.invoke({"user_id": "abc"}))
# Expected output: Error: user_id must be numeric, I got 'abc'

Pattern: the defensive tool wrapper

When you have many tools, you can build a wrapper that applies try/except automatically:

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool

def safe_tool_execute(tool_fn, args: dict) -> str:
    """Run a tool with standard error handling."""
    try:
        result = tool_fn.invoke(args)
        if result is None:
            return "The tool returned no result."
        return str(result)
    except TypeError as e:
        return f"Argument error: {e}"
    except TimeoutError:
        return "Error: the operation took too long. Try again."
    except Exception as e:
        return f"Error running {tool_fn.name}: {type(e).__name__}: {e}"

@tool
def divide(a: float, b: float) -> str:
    """Divide two numbers."""
    return str(a / b)

print(safe_tool_execute(divide, {"a": 10, "b": 3}))
# Expected output: 3.3333333333333335

print(safe_tool_execute(divide, {"a": 10, "b": 0}))
# Expected output: Error running divide: ZeroDivisionError: float division by zero

Retry logic with exponential backoff

Transient errors — network timeouts, temporary rate limits, 503s from the server — fix themselves if you wait a moment and try again. But retrying immediately usually makes things worse (more load on an already overloaded server). The standard pattern is exponential backoff: wait 1s, then 2s, then 4s.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool
import time
import random

def execute_with_retry(tool_fn, args: dict, max_retries: int = 3, base_delay: float = 1.0) -> str:
    """Run a tool with retry and exponential backoff."""
    for attempt in range(max_retries):
        try:
            result = tool_fn.invoke(args)
            return result
        except Exception as e:
            is_last_attempt = attempt == max_retries - 1
            if is_last_attempt:
                return f"Error after {max_retries} attempts: {type(e).__name__}: {e}"

            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            print(f"  ⚠️ Attempt {attempt + 1} failed: {e}. Retrying in {delay:.1f}s...")
            time.sleep(delay)

    return "Error: retries exhausted"

@tool
def unreliable_api(query: str) -> str:
    """Simulate an API that fails intermittently."""
    if random.random() < 0.7:
        raise ConnectionError("Server unavailable")
    return f"Result for: {query}"

result = execute_with_retry(unreliable_api, {"query": "test"})
print(f"Result: {result}")
# Expected output (varies because of the randomness):
#   ⚠️ Attempt 1 failed: Server unavailable. Retrying in 1.3s...
#   ⚠️ Attempt 2 failed: Server unavailable. Retrying in 2.2s...
# Result: Result for: test

Why add jitter (random noise)?

The random.uniform(0, 0.5) adds a small random delay to every retry. Without jitter, if 100 clients fail at the same time and all retry at exactly 1s, the server takes another 100 simultaneous requests. With jitter, the requests spread out over a time window, giving the server room to recover.

Selective retry: transient errors only

Not every error deserves a retry. A "ticker not found" error won't fix itself no matter how many times you try. You should only retry transient errors:

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool, ToolException
import time
import random

TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)

def execute_with_smart_retry(tool_fn, args: dict, max_retries: int = 3) -> str:
    """Retry only for transient errors. Permanent errors fail immediately."""
    for attempt in range(max_retries):
        try:
            return tool_fn.invoke(args)
        except TRANSIENT_ERRORS as e:
            if attempt == max_retries - 1:
                return f"Persistent transient error after {max_retries} attempts: {e}"
            delay = (2 ** attempt) + random.uniform(0, 0.5)
            print(f"  ⚠️ Transient error (attempt {attempt + 1}): {e}. Retry in {delay:.1f}s...")
            time.sleep(delay)
        except ToolException as e:
            return f"Tool error (not retryable): {e}"
        except Exception as e:
            return f"Permanent error: {type(e).__name__}: {e}"

    return "Error: retries exhausted"

Graceful degradation: what to do when a tool fails permanently

Sometimes the tool simply isn't going to work — the API has been down for hours, the key expired, or the service no longer exists. In those cases you need a graceful degradation strategy: give the user something useful even though the tool is broken.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool

@tool
def get_weather_with_fallback(city: str) -> str:
    """Get the current weather for a city."""
    try:
        raise ConnectionError("Weather API unavailable")
    except ConnectionError:
        return (
            f"⚠️ I couldn't get real-time weather for {city}. "
            f"The weather service is unavailable right now. "
            f"You can check it directly at https://weather.com"
        )

result = get_weather_with_fallback.invoke({"city": "Madrid"})
print(result)
# Expected output:
# ⚠️ I couldn't get real-time weather for Madrid. The weather service is
# unavailable right now. You can check it directly at
# https://weather.com

Comparison: fail-fast vs retry vs fallback

StrategyWhen to use itBehavior
Fail-fastValidation errors, invalid argumentsReturns the error immediately, no retry
RetryTransient errors (network, timeout, 503)Retries with exponential backoff
FallbackTool permanently unavailableReturns a useful alternative answer
from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool
import time
import random

@tool
def robust_weather(city: str) -> str:
    """Get the weather with a full error handling strategy."""
    if not city or len(city) < 2:
        return f"Error: invalid city '{city}'. Provide a valid city name."

    weather_data = {"Madrid": "22°C, sunny", "London": "14°C, cloudy", "Tokyo": "28°C, humid"}

    max_retries = 3
    for attempt in range(max_retries):
        try:
            if random.random() < 0.3:
                raise ConnectionError("Network timeout")

            if city in weather_data:
                return f"Weather in {city}: {weather_data[city]}"
            else:
                return f"I have no weather data for {city}. Available cities: {', '.join(weather_data.keys())}"

        except ConnectionError as e:
            if attempt < max_retries - 1:
                time.sleep(0.1 * (2 ** attempt))
                continue
            return f"⚠️ I couldn't reach the weather service after {max_retries} attempts. Try again later."

result = robust_weather.invoke({"city": "Madrid"})
print(result)
# Expected output: Weather in Madrid: 22°C, sunny

ToolException for structured error reporting

LangChain provides ToolException as a structured way to report errors inside tools. When a tool raises ToolException, LangChain catches it and turns it into a ToolMessage carrying the error — which lets the model see the failure and decide what to do (retry, ask for clarification, or use a different tool).

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool, ToolException

@tool(handle_tool_error=True)
def get_stock_price(ticker: str) -> str:
    """Get the current price of a stock."""
    prices = {"AAPL": 185.50, "GOOGL": 142.30, "MSFT": 420.80}

    ticker = ticker.upper().strip()
    if ticker not in prices:
        raise ToolException(
            f"Ticker '{ticker}' not found. "
            f"Available tickers: {', '.join(sorted(prices.keys()))}. "
            f"Check the symbol and try again."
        )

    return f"{ticker}: ${prices[ticker]:.2f} USD"

print(get_stock_price.invoke({"ticker": "AAPL"}))
# Expected output: AAPL: $185.50 USD

print(get_stock_price.invoke({"ticker": "INVALID"}))
# Expected output: Ticker 'INVALID' not found. Available tickers: AAPL, GOOGL, MSFT.
# Check the symbol and try again.

handle_tool_error: controlling how the error is reported

The handle_tool_error parameter on @tool controls what happens when the tool raises ToolException:

from langchain_core.tools import tool, ToolException

# Option 1: handle_tool_error=True → returns str(exception) as a ToolMessage
@tool(handle_tool_error=True)
def tool_v1(x: str) -> str:
    """Demo tool."""
    raise ToolException("Something failed")

# Option 2: handle_tool_error="custom message" → returns that message
@tool(handle_tool_error="The tool is unavailable. Try rephrasing your question.")
def tool_v2(x: str) -> str:
    """Demo tool."""
    raise ToolException("Internal error")

# Option 3: handle_tool_error=callable → processes the error with a function
def custom_handler(error: ToolException) -> str:
    return f"⚠️ Handled error: {error}. Please try different parameters."

@tool(handle_tool_error=custom_handler)
def tool_v3(x: str) -> str:
    """Demo tool."""
    raise ToolException("Data unavailable")

print(tool_v1.invoke({"x": "test"}))
# Expected output: Something failed

print(tool_v2.invoke({"x": "test"}))
# Expected output: The tool is unavailable. Try rephrasing your question.

print(tool_v3.invoke({"x": "test"}))
# Expected output: ⚠️ Handled error: Data unavailable. Please try different parameters.

ToolException vs returning an error string

When should you use ToolException and when should you simply return a string with the error?

Aspectreturn "Error: ..."raise ToolException(...)
ControlAlways goes back to the modelOnly if handle_tool_error is configured
LoggingIndistinguishable from a normal resultCan be intercepted in middleware/callbacks
SemanticsThe model can't tell error from resultA clear signal that something failed
ProductionGood enough for prototypesPreferred for monitored systems

Rule of thumb: Use return "Error: ..." during development. Move to ToolException when you need monitoring and logging that distinguishes successful results from errors.


Debugging tool calls: inspecting messages

When something doesn't work — the model doesn't call the tool you expect, it passes the wrong arguments, or the result doesn't get integrated properly — you need to inspect the messages. Tool calls live inside the AIMessage the model generates.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return f"Error: the expression contains disallowed characters"
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([calculate])

response = model_with_tools.invoke("What's 15% of 240?")

print("=== AIMessage inspection ===")
print(f"Content: '{response.content}'")
print(f"Tool calls: {response.tool_calls}")
print(f"Number of tool calls: {len(response.tool_calls)}")

if response.tool_calls:
    tc = response.tool_calls[0]
    print(f"\n=== Tool call detail ===")
    print(f"  Name: {tc['name']}")
    print(f"  Args: {tc['args']}")
    print(f"  ID:   {tc['id']}")

# Expected output:
# === AIMessage inspection ===
# Content: ''
# Tool calls: [{'name': 'calculate', 'args': {'expression': '240 * 0.15'}, 'id': 'call_abc123'}]
# Number of tool calls: 1
#
# === Tool call detail ===
#   Name: calculate
#   Args: {'expression': '240 * 0.15'}
#   ID:   call_abc123

Inspecting the full flow

To debug the whole loop (model → tool → model), print every message in the conversation:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        return f"Error: the expression contains disallowed characters"
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([calculate])

messages = [HumanMessage(content="What's 15% of 240?")]

response = model_with_tools.invoke(messages)
messages.append(response)

if response.tool_calls:
    for tc in response.tool_calls:
        result = calculate.invoke(tc["args"])
        messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

    final = model_with_tools.invoke(messages)
    messages.append(final)

print("=== Full message flow ===")
for i, msg in enumerate(messages):
    msg_type = type(msg).__name__
    if hasattr(msg, "tool_calls") and msg.tool_calls:
        tools = [tc["name"] for tc in msg.tool_calls]
        print(f"  [{i}] {msg_type}: tool_calls={tools}")
    elif hasattr(msg, "tool_call_id"):
        print(f"  [{i}] {msg_type}: '{msg.content}' (id={msg.tool_call_id})")
    else:
        print(f"  [{i}] {msg_type}: '{msg.content[:80]}'")

# Expected output:
# === Full message flow ===
#   [0] HumanMessage: 'What's 15% of 240?'
#   [1] AIMessage: tool_calls=['calculate']
#   [2] ToolMessage: '36.0' (id=call_abc123)
#   [3] AIMessage: '15% of 240 is 36.'

A tool execution loop with error handling built in

The tool execution loop you learned in capsule 04 assumed tools always work. Here's the robust version that handles errors at every step:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool, ToolException
from langchain_core.messages import HumanMessage, ToolMessage
import time
import random

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weather = {"Madrid": "22°C, sunny", "London": "14°C, cloudy", "Tokyo": "28°C, humid"}
    city_normalized = city.strip().title()
    if city_normalized not in weather:
        raise ToolException(f"City '{city}' not found. Available: {', '.join(weather.keys())}")
    return f"Weather in {city_normalized}: {weather[city_normalized]}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a simple math expression."""
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        raise ToolException(f"Invalid expression: disallowed characters in '{expression}'")
    try:
        return str(eval(expression))
    except Exception as e:
        raise ToolException(f"I couldn't evaluate '{expression}': {e}")

tools = [get_weather, calculate]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

def execute_tool_call(tool_call: dict, max_retries: int = 2) -> str:
    """Run a single tool call with validation and retry."""
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]
    tool_id = tool_call["id"]

    if tool_name not in tools_by_name:
        return f"Error: tool '{tool_name}' does not exist. Available tools: {', '.join(tools_by_name.keys())}"

    selected_tool = tools_by_name[tool_name]

    for attempt in range(max_retries):
        try:
            result = selected_tool.invoke(tool_args)
            return result
        except ToolException as e:
            return str(e)
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(0.5 * (2 ** attempt))
                continue
            return f"Error after {max_retries} attempts on '{tool_name}': {e}"

    return f"Unexpected error on '{tool_name}'"

def run_tool_loop(question: str, max_iterations: int = 5) -> str:
    """Run the full model→tool→model loop with error handling."""
    messages = [HumanMessage(content=question)]

    for iteration in range(max_iterations):
        response = model_with_tools.invoke(messages)
        messages.append(response)

        if not response.tool_calls:
            return response.content

        print(f"  Iteration {iteration + 1}: {len(response.tool_calls)} tool call(s)")

        for tc in response.tool_calls:
            result = execute_tool_call(tc)
            print(f"    → {tc['name']}({tc['args']}) = {result[:60]}")
            messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

    return "Iteration limit reached. The model never produced a final answer."

answer = run_tool_loop("What's 100 * 1.16 and what's the weather in Madrid?")
print(f"\nAnswer: {answer}")
# Expected output:
#   Iteration 1: 2 tool call(s)
#     → calculate({'expression': '100 * 1.16'}) = 116.0
#     → get_weather({'city': 'Madrid'}) = Weather in Madrid: 22°C, sunny
#
# Answer: 100 × 1.16 = 116.0. In Madrid it's 22°C and sunny.

Connection to the project

In this module's project (Capsule 08), you'll build an assistant with external tools that pulls together everything you learned about error handling:

  • Argument validation: Every tool in the assistant (weather, calculator, web search) validates its inputs before running
  • Defensive try/except: Every tool wraps its logic in try/except so one failure doesn't take down the system
  • Graceful degradation: When a tool fails, the assistant tells the user and keeps working with the remaining tools
  • A robust tool execution loop: The loop you'll implement handles nonexistent tools, execution errors, and iteration limits

The patterns in this capsule are what make your assistant reliable, not just functional.


Troubleshooting

Problem 1: the model calls a tool that doesn't exist

KeyError: 'web_search'

Cause: The model generates a tool_call with a name that isn't in your tools dictionary. This happens when the model "hallucinates" a tool name.

Solution: Always check that the name exists before executing:

if tool_name not in tools_by_name:
    error_msg = f"Tool '{tool_name}' does not exist. Available: {list(tools_by_name.keys())}"
    messages.append(ToolMessage(content=error_msg, tool_call_id=tool_call["id"]))

By returning the error as a ToolMessage, the model sees it and can correct itself on the next iteration.

Problem 2: ToolException isn't caught and crashes the program

langchain_core.tools.base.ToolException: Ticker 'XYZ' not found

Cause: You raise ToolException but you didn't set handle_tool_error=True on the @tool decorator, and you don't catch it in your try/except either.

Solution: Add handle_tool_error=True to the decorator or catch ToolException explicitly:

@tool(handle_tool_error=True)
def my_tool(x: str) -> str:
    """My tool."""
    raise ToolException("Handled error")

# Or catch it in your loop:
try:
    result = tool_fn.invoke(args)
except ToolException as e:
    result = str(e)

Problem 3: the retry loop doesn't wait between attempts

Cause: You're using time.sleep(0) or the delay doesn't scale. Check that the backoff math is right: delay = base * (2 ** attempt) gives 1s, 2s, 4s when base=1.

Solution:

delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
print(f"Waiting {delay:.1f}s before retrying...")
time.sleep(delay)

Problem 4: a ToolMessage without tool_call_id raises an error

ValueError: ToolMessage must have a tool_call_id

Cause: When you built the ToolMessage, you didn't pass the tool_call_id that came with the original tool_call.

Solution: Always pass the tool call's ID:

for tc in response.tool_calls:
    result = execute_tool(tc)
    messages.append(ToolMessage(
        content=result,
        tool_call_id=tc["id"]    # ← required
    ))

Problem 5: the model gets stuck in an infinite tool call loop

Cause: The model keeps calling tools without producing a final answer. This happens when the tool's response doesn't satisfy the model and it keeps trying.

Solution: Cap the loop's iterations and add a cutoff message:

def run_tool_loop(question, max_iterations=5):
    for iteration in range(max_iterations):
        response = model_with_tools.invoke(messages)
        if not response.tool_calls:
            return response.content
        # ... run tools ...
    return "Iteration limit reached."

Problem 6: tool errors never reach the model

Cause: You catch the exception but you don't send it back as a ToolMessage. The model expects a response for every tool call; if it doesn't get one, the flow breaks.

Solution: Always return one ToolMessage per tool call, even when it failed:

try:
    result = tool_fn.invoke(args)
except Exception as e:
    result = f"Error: {e}"

messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

Problem 7: handle_tool_error with a custom function doesn't work

Cause: Your custom function doesn't have the right signature. It must take a ToolException and return a str.

Solution:

def my_handler(error: ToolException) -> str:
    return f"Handled error: {error}"

@tool(handle_tool_error=my_handler)
def my_tool(x: str) -> str:
    """My tool."""
    raise ToolException("something failed")

Exercises

Exercise 1: Defensive argument validation (Easy)

Build a tool convert_temperature that converts between Celsius and Fahrenheit. It must validate that the scale is "C" or "F", that the temperature is a reasonable number (-100 to 1000), and return clear error messages.

See solution
from langchain_core.tools import tool

@tool
def convert_temperature(value: float, from_scale: str) -> str:
    """Convert a temperature between Celsius and Fahrenheit."""
    from_scale = from_scale.upper().strip()

    if from_scale not in ("C", "F"):
        return f"Error: scale '{from_scale}' is not valid. Use 'C' (Celsius) or 'F' (Fahrenheit)."

    if not -100 <= value <= 1000:
        return f"Error: temperature {value} is outside the reasonable range (-100 to 1000)."

    if from_scale == "C":
        result = (value * 9/5) + 32
        return f"{value}°C = {result:.1f}°F"
    else:
        result = (value - 32) * 5/9
        return f"{value}°F = {result:.1f}°C"

print(convert_temperature.invoke({"value": 100, "from_scale": "C"}))
print(convert_temperature.invoke({"value": 72, "from_scale": "F"}))
print(convert_temperature.invoke({"value": 25, "from_scale": "K"}))
print(convert_temperature.invoke({"value": 5000, "from_scale": "C"}))
# Expected output:
# 100°C = 212.0°F
# 72°F = 22.2°C
# Error: scale 'K' is not valid. Use 'C' (Celsius) or 'F' (Fahrenheit).
# Error: temperature 5000 is outside the reasonable range (-100 to 1000).

Explanation: Every possible error produces a descriptive message the model can use to correct itself or to inform the user. Normalizing with .upper().strip() handles variations like "c", " C ", and so on.

Exercise 2: A tool with ToolException and a custom handler (Easy)

Build a tool lookup_country that looks up information about a country by name. Use ToolException for errors and a custom handler that formats the error in a friendly way.

See solution
from langchain_core.tools import tool, ToolException

def friendly_error_handler(error: ToolException) -> str:
    return f"🔍 I couldn't complete the lookup: {error}. Try another country."

@tool(handle_tool_error=friendly_error_handler)
def lookup_country(country: str) -> str:
    """Look up basic information about a country."""
    countries = {
        "Mexico": {"capital": "Mexico City", "population": "130M", "language": "Spanish"},
        "Spain": {"capital": "Madrid", "population": "47M", "language": "Spanish"},
        "Japan": {"capital": "Tokyo", "population": "125M", "language": "Japanese"},
    }

    country_normalized = country.strip().title()
    if country_normalized not in countries:
        available = ", ".join(sorted(countries.keys()))
        raise ToolException(f"Country '{country}' not found. Available: {available}")

    info = countries[country_normalized]
    return f"{country_normalized} — Capital: {info['capital']}, Population: {info['population']}, Language: {info['language']}"

print(lookup_country.invoke({"country": "Mexico"}))
print(lookup_country.invoke({"country": "France"}))
# Expected output:
# Mexico — Capital: Mexico City, Population: 130M, Language: Spanish
# 🔍 I couldn't complete the lookup: Country 'France' not found. Available: Japan, Mexico, Spain. Try another country.

Explanation: ToolException signals a semantic error (country not found). The custom handler turns that error into a friendly message with an emoji and a suggestion. That's what the model receives as a ToolMessage.

Exercise 3: Retry with backoff for an unstable API (Medium)

Build a function resilient_invoke that runs any tool with retry and exponential backoff. It must accept max_retries and base_delay, print each attempt, and distinguish transient errors (retry) from permanent errors (fail-fast).

See solution
from langchain_core.tools import tool, ToolException
import time
import random

TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)

def resilient_invoke(tool_fn, args: dict, max_retries: int = 3, base_delay: float = 1.0) -> dict:
    """Run a tool with smart retry.
    Returns a dict with the result, the attempts, and the total time.
    """
    start = time.time()

    for attempt in range(max_retries):
        try:
            result = tool_fn.invoke(args)
            return {
                "success": True,
                "result": result,
                "attempts": attempt + 1,
                "elapsed_ms": (time.time() - start) * 1000,
            }
        except TRANSIENT_ERRORS as e:
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 0.3)
                print(f"  ⚠️ Attempt {attempt + 1}/{max_retries}: {type(e).__name__}. Retry in {delay:.1f}s...")
                time.sleep(delay)
            else:
                return {
                    "success": False,
                    "result": f"Persistent transient error: {e}",
                    "attempts": max_retries,
                    "elapsed_ms": (time.time() - start) * 1000,
                }
        except ToolException as e:
            return {
                "success": False,
                "result": f"Tool error: {e}",
                "attempts": attempt + 1,
                "elapsed_ms": (time.time() - start) * 1000,
            }
        except Exception as e:
            return {
                "success": False,
                "result": f"Permanent error: {type(e).__name__}: {e}",
                "attempts": attempt + 1,
                "elapsed_ms": (time.time() - start) * 1000,
            }

call_count = 0

@tool
def flaky_api(query: str) -> str:
    """Simulate an API that fails the first 2 times."""
    global call_count
    call_count += 1
    if call_count <= 2:
        raise ConnectionError(f"Timeout (internal attempt #{call_count})")
    return f"Successful result for: {query}"

call_count = 0
result = resilient_invoke(flaky_api, {"query": "test"}, max_retries=3, base_delay=0.1)
print(f"\nResult: {result}")
# Expected output:
#   ⚠️ Attempt 1/3: ConnectionError. Retry in 0.1s...
#   ⚠️ Attempt 2/3: ConnectionError. Retry in 0.3s...
#
# Result: {'success': True, 'result': 'Successful result for: test', 'attempts': 3, 'elapsed_ms': 452.3}

Explanation: The function distinguishes three kinds of errors: transient ones (retry with backoff), ToolException (fail-fast, a business error), and generic ones (fail-fast, an unexpected error). The returned dict gives you visibility into attempts and timing — useful for monitoring.

Exercise 4: A debug inspector for tool calls (Medium)

Build a function debug_tool_loop that runs the full model→tool→model loop and prints a detailed report of every step: message type, content, tool calls, and timings.

See solution
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
import time

@tool
def multiply(a: float, b: float) -> str:
    """Multiply two numbers."""
    return str(a * b)

@tool
def add(a: float, b: float) -> str:
    """Add two numbers."""
    return str(a + b)

tools = [multiply, add]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

def debug_tool_loop(question: str, max_iterations: int = 5) -> str:
    """A tool loop with detailed debug output at every step."""
    messages = [HumanMessage(content=question)]
    print(f"{'=' * 60}")
    print(f"DEBUG TOOL LOOP")
    print(f"Question: {question}")
    print(f"Available tools: {list(tools_by_name.keys())}")
    print(f"{'=' * 60}")

    for iteration in range(max_iterations):
        print(f"\n--- Iteration {iteration + 1} ---")

        start = time.time()
        response = model_with_tools.invoke(messages)
        model_time = (time.time() - start) * 1000

        print(f"  [MODEL] ({model_time:.0f}ms)")
        print(f"    Content: '{response.content[:100]}'" if response.content else "    Content: (empty)")
        print(f"    Tool calls: {len(response.tool_calls)}")

        messages.append(response)

        if not response.tool_calls:
            print(f"\n{'=' * 60}")
            print(f"FINAL RESULT: {response.content[:200]}")
            print(f"{'=' * 60}")
            return response.content

        for i, tc in enumerate(response.tool_calls):
            print(f"\n  [TOOL CALL {i + 1}]")
            print(f"    Name: {tc['name']}")
            print(f"    Args: {tc['args']}")
            print(f"    ID:   {tc['id']}")

            start = time.time()
            if tc["name"] in tools_by_name:
                try:
                    result = tools_by_name[tc["name"]].invoke(tc["args"])
                except Exception as e:
                    result = f"Error: {e}"
            else:
                result = f"Error: tool '{tc['name']}' does not exist"
            tool_time = (time.time() - start) * 1000

            print(f"    Result: {result}")
            print(f"    Time: {tool_time:.0f}ms")

            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

    return "Iteration limit reached"

result = debug_tool_loop("What's 7 * 8 + 3 * 4?")
# Expected output:
# ============================================================
# DEBUG TOOL LOOP
# Question: What's 7 * 8 + 3 * 4?
# Available tools: ['multiply', 'add']
# ============================================================
#
# --- Iteration 1 ---
#   [MODEL] (823ms)
#     Content: (empty)
#     Tool calls: 2
#
#   [TOOL CALL 1]
#     Name: multiply
#     Args: {'a': 7.0, 'b': 8.0}
#     ID:   call_abc123
#     Result: 56.0
#     Time: 0ms
#
#   [TOOL CALL 2]
#     Name: multiply
#     Args: {'a': 3.0, 'b': 4.0}
#     ID:   call_def456
#     Result: 12.0
#     Time: 0ms
#
# --- Iteration 2 ---
#   [MODEL] (645ms)
#     Content: (empty)
#     Tool calls: 1
#
#   [TOOL CALL 1]
#     Name: add
#     Args: {'a': 56.0, 'b': 12.0}
#     ID:   call_ghi789
#     Result: 68.0
#     Time: 0ms
#
# --- Iteration 3 ---
#   [MODEL] (412ms)
#     Content: '7 × 8 + 3 × 4 = 68'
#     Tool calls: 0
#
# ============================================================
# FINAL RESULT: 7 × 8 + 3 × 4 = 68
# ============================================================

Explanation: This debug inspector shows exactly what the model decides at each step: which tools it calls, with which arguments, how long each operation takes, and when it decides to produce the final answer. It's invaluable when a tool call doesn't behave the way you expect.

Exercise 5: A tool system with fallback across sources (Hard)

Build a tool smart_search that looks up information across three (simulated) "sources." If the first source fails, try the second. If the second fails, try the third. Return the result from the first source that works, along with metadata about which source answered.

See solution
from langchain_core.tools import tool
import random

def source_api_primary(query: str) -> str:
    """Primary source — fails 50% of the time."""
    if random.random() < 0.5:
        raise ConnectionError("Primary API unavailable")
    return f"[Primary source] Result for '{query}': detailed, up-to-date information."

def source_api_secondary(query: str) -> str:
    """Secondary source — fails 30% of the time."""
    if random.random() < 0.3:
        raise ConnectionError("Secondary API unavailable")
    return f"[Secondary source] Result for '{query}': general information."

def source_cache(query: str) -> str:
    """Local cache — never fails, but the data may be stale."""
    return f"[Local cache] Result for '{query}': data from 24 hours ago."

@tool
def smart_search(query: str) -> str:
    """Search for information across multiple sources with automatic fallback."""
    sources = [
        ("Primary API", source_api_primary),
        ("Secondary API", source_api_secondary),
        ("Local Cache", source_cache),
    ]

    errors = []

    for source_name, source_fn in sources:
        try:
            result = source_fn(query)
            if errors:
                fallback_info = f" (after failures in: {', '.join(errors)})"
            else:
                fallback_info = ""
            return f"{result}{fallback_info}"
        except Exception as e:
            errors.append(source_name)
            continue

    return f"Error: every source failed — {', '.join(errors)}"

random.seed(42)
for i in range(5):
    result = smart_search.invoke({"query": f"Python tutorial"})
    print(f"Search {i + 1}: {result}")
    print()
# Expected output (varies with the seed):
# Search 1: [Primary source] Result for 'Python tutorial': detailed, up-to-date information.
#
# Search 2: [Secondary source] Result for 'Python tutorial': general information. (after failures in: Primary API)
#
# Search 3: [Local cache] Result for 'Python tutorial': data from 24 hours ago. (after failures in: Primary API, Secondary API)
# ...

Explanation: The fallback-across-sources pattern is identical to the one you saw in the Module 1 project (fallback across LLM providers), just applied to tools. The "after failures in..." metadata gives you visibility into the path the request took — useful for debugging and monitoring.

Exercise 6: A complete tool execution loop with error handling (Hard)

Implement a function safe_agent_loop that runs the model→tool→model loop with: tool name validation, try/except per tool call, retry for transient errors, an iteration limit, and an execution summary at the end (tools called, errors, total time).

See solution
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool, ToolException
from langchain_core.messages import HumanMessage, ToolMessage
import time
import random

@tool
def get_price(product: str) -> str:
    """Get the price of a product."""
    prices = {"laptop": 999.99, "mouse": 29.99, "keyboard": 79.99, "monitor": 349.99}
    product = product.lower().strip()
    if product not in prices:
        raise ToolException(f"Product '{product}' not found. Available: {', '.join(prices.keys())}")
    return f"{product}: ${prices[product]}"

@tool
def calculate_discount(price: float, percent: float) -> str:
    """Calculate the discounted price."""
    if percent < 0 or percent > 100:
        raise ToolException(f"Percentage {percent} is invalid. It must be between 0 and 100.")
    discount = price * (percent / 100)
    final = price - discount
    return f"Original price: ${price:.2f}, Discount: ${discount:.2f} ({percent}%), Final price: ${final:.2f}"

tools = [get_price, calculate_discount]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

def safe_agent_loop(question: str, max_iterations: int = 5, max_retries: int = 2) -> dict:
    """The complete loop with error handling, retry, and metrics."""
    messages = [HumanMessage(content=question)]
    execution_log = []
    start_time = time.time()

    for iteration in range(max_iterations):
        response = model_with_tools.invoke(messages)
        messages.append(response)

        if not response.tool_calls:
            elapsed = (time.time() - start_time) * 1000
            return {
                "answer": response.content,
                "iterations": iteration + 1,
                "tool_calls": len(execution_log),
                "errors": sum(1 for log in execution_log if not log["success"]),
                "elapsed_ms": elapsed,
                "log": execution_log,
            }

        for tc in response.tool_calls:
            log_entry = {"tool": tc["name"], "args": tc["args"], "success": False, "result": ""}

            if tc["name"] not in tools_by_name:
                log_entry["result"] = f"Tool '{tc['name']}' does not exist"
                execution_log.append(log_entry)
                messages.append(ToolMessage(
                    content=f"Error: tool '{tc['name']}' unavailable. Valid tools: {', '.join(tools_by_name.keys())}",
                    tool_call_id=tc["id"],
                ))
                continue

            result = None
            for attempt in range(max_retries):
                try:
                    result = tools_by_name[tc["name"]].invoke(tc["args"])
                    log_entry["success"] = True
                    log_entry["result"] = result
                    break
                except ToolException as e:
                    log_entry["result"] = str(e)
                    result = str(e)
                    break
                except Exception as e:
                    if attempt < max_retries - 1:
                        time.sleep(0.1 * (2 ** attempt))
                    else:
                        log_entry["result"] = f"Error after {max_retries} attempts: {e}"
                        result = log_entry["result"]

            execution_log.append(log_entry)
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

    elapsed = (time.time() - start_time) * 1000
    return {
        "answer": "Iteration limit reached.",
        "iterations": max_iterations,
        "tool_calls": len(execution_log),
        "errors": sum(1 for log in execution_log if not log["success"]),
        "elapsed_ms": elapsed,
        "log": execution_log,
    }

result = safe_agent_loop("How much is a laptop with a 15% discount?")

print(f"Answer: {result['answer']}")
print(f"Iterations: {result['iterations']}")
print(f"Tool calls: {result['tool_calls']}")
print(f"Errors: {result['errors']}")
print(f"Time: {result['elapsed_ms']:.0f}ms")
print(f"\nExecution log:")
for log in result["log"]:
    status = "✅" if log["success"] else "❌"
    print(f"  {status} {log['tool']}({log['args']}) → {log['result'][:60]}")
# Expected output:
# Answer: A laptop costs $999.99. With a 15% discount, the final price is $849.99.
# Iterations: 3
# Tool calls: 2
# Errors: 0
# Time: 2845ms
#
# Execution log:
#   ✅ get_price({'product': 'laptop'}) → laptop: $999.99
#   ✅ calculate_discount({'price': 999.99, 'percent': 15.0}) → Original price: $999.99, Discount: $150.00 (15.0%), Fin

Explanation: This function combines every pattern in the capsule: tool name validation, try/except with retry, graceful degradation, an iteration limit, and a detailed execution log. The returned dict gives you full visibility into what happened — how many iterations, how many errors, which tools were called and with what result.


Summary

In this capsule you learned:

  • The 5 most common errors in tool calling: tool not found, invalid arguments, timeout, unexpected format, rate limiting
  • Validating arguments before execution avoids unnecessary errors and gives the model clear messages
  • Try/except wraps any execution that touches external services
  • Retry with exponential backoff handles transient errors: delay = base * (2 ** attempt)
  • Jitter (random noise) spreads retries out so you don't overload servers
  • Graceful degradation gives useful answers when a tool fails permanently
  • ToolException reports errors in a structured way; handle_tool_error controls how they're presented
  • Inspecting messages (tool_calls, content, ToolMessage) is how you debug tool calling flows
  • Only retry transient errors (network, timeout); business errors should fail immediately

Next capsule: Project — you'll build an assistant with external tools that brings together everything you've learned: tools with @tool, bind_tools, the tool execution loop, parallel calls, and the robust error handling from this capsule.


Further reading

  1. LangChain Tool Error Handling — The official guide to handling errors in tools
  2. ToolException API Reference — Reference for the ToolException class
  3. LangChain Tools How-To — Index of how-to guides for tools
  4. Exponential Backoff and Jitter (AWS) — The classic AWS article on backoff strategies
  5. LangChain Tool Calling — Tool calling concepts in LangChain
  6. Python Exception Handling Best Practices — The official Python reference on exception handling

Module 2 — LangChain & LangGraph: From Chains to Agents