6. Tool Validation and Error Handling

Capsule overview

So far you've built tools that assume everything goes well: the model passes correct arguments, the API responds, the network works, and the result is valid. In production, that's a disaster waiting to happen. APIs go down, timeouts happen, rate limits arrive without warning, and models — however good they are — sometimes pass arguments that make no sense. An agent without error handling is a bug waiting to happen.

This capsule shifts your mindset: error handling isn't a "nice to have" you bolt on at the end. It's a first-class concern that goes into every tool from minute zero. You'll learn to validate inputs with Pydantic before your code runs, to catch errors without crashing the agent, to implement retry logic with exponential backoff, and to design graceful degradation so your agent keeps working when a tool fails.

Why can the model self-correct? Because when you return "Error: city cannot be empty" as a ToolMessage, the model receives that text in its context, understands something went wrong, and on the next iteration it can try different arguments, use another tool, or tell the user. If the exception propagates and crashes the loop, the model never gets any feedback — the agent simply dies. That difference is what separates a demo agent from a production-ready one.

Input Validation with Pydantic

Why validate before executing

When the model calls a tool, it passes arguments based on the JSON schema. Most of the time they're correct. But "most of the time" isn't good enough for production. The model can pass empty strings, out-of-range numbers, or unexpected formats.

Pydantic validates before execution. If the model passes city: "" to a tool expecting min_length=1, Pydantic raises a ValidationError before your function ever runs.

Field constraints

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

class WeatherInput(BaseModel):
    city: str = Field(
        min_length=1,
        max_length=100,
        description="City name. Cannot be empty."
    )
    units: str = Field(
        default="celsius",
        pattern="^(celsius|fahrenheit)$",
        description="Units: 'celsius' or 'fahrenheit'."
    )

@tool(args_schema=WeatherInput)
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city."""
    temps = {"Madrid": 22, "Tokyo": 18, "NYC": 15}
    temp = temps.get(city, 20)
    if units == "fahrenheit":
        temp = temp * 9/5 + 32
    symbol = "°F" if units == "fahrenheit" else "°C"
    return f"{city}: {temp}{symbol}"

If the model passes city: "", Pydantic rejects it before get_weather runs. If it passes units: "kelvin", the pattern catches it.

Constraints reference

Custom validators

For validations the basic constraints don't cover:

from pydantic import BaseModel, Field, field_validator, model_validator

class CalculatorInput(BaseModel):
    expression: str = Field(description="A safe math expression")

    @field_validator("expression")
    @classmethod
    def validate_safe_expression(cls, v: str) -> str:
        import re
        if not re.match(r'^[\d\s\+\-\*\/\(\)\.]+$', v):
            raise ValueError(f"Disallowed characters: '{v}'. Only: numbers, +, -, *, /, (), .")
        return v.strip()


class DateRangeInput(BaseModel):
    start_date: str = Field(description="Start date YYYY-MM-DD")
    end_date: str = Field(description="End date YYYY-MM-DD")

    @model_validator(mode="after")
    def validate_date_range(self):
        from datetime import datetime
        start = datetime.strptime(self.start_date, "%Y-%m-%d")
        end = datetime.strptime(self.end_date, "%Y-%m-%d")
        if end < start:
            raise ValueError(f"end_date cannot be earlier than start_date")
        return self

field_validator validates a single field. model_validator(mode="after") validates combinations of fields. If they raise ValueError, it becomes a ValidationError with a descriptive message the model can read.

Error Handling Inside Tools

The golden rule

Inside a tool, never let an exception escape the function. Always return a string with the error.

If an exception escapes the tool, it crashes the tool execution loop. The model never gets feedback. The agent dies. If instead you return a string like "Error: API timeout after 10s", that arrives as a ToolMessage and the model can: (1) retry with different arguments, (2) use an alternative tool, or (3) tell the user.

The basic pattern

import requests
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    try:
        response = requests.get(
            "https://api.example.com/search",
            params={"q": query},
            timeout=10
        )
        response.raise_for_status()
        return response.json()["results"][0]["snippet"]
    except requests.Timeout:
        return "Error: The search exceeded the 10s timeout. Try a shorter query."
    except requests.HTTPError as e:
        return f"Error: The API returned status {e.response.status_code}."
    except requests.ConnectionError:
        return "Error: Could not connect to the search service."
    except (KeyError, IndexError):
        return "Error: The search returned no valid results."
    except Exception as e:
        return f"Unexpected error: {type(e).__name__}: {str(e)}"

Exception hierarchy: from the most specific (Timeout) to the most general (Exception). Every message is actionable.

Effective error messages

A good error message has three components: what failed, why, and what to do.

# ❌ Bad — the model doesn't know what to do
return f"Error: {e}"

# ❌ Bad — too technical
return f"requests.exceptions.ConnectionError: [Errno -2] Name or service not known"

# ✅ Good — clear, specific, actionable
return "Error: Could not connect to the weather service. Try another city or tell the user."

# ✅ Good — offers alternatives
return "Error: Search failed with a timeout. Options: retry with a shorter query, or use information you already have."

The model self-corrects

When the model receives a descriptive error, it can correct itself:

@tool
def get_city_weather(city: str) -> str:
    """Get the current weather for a city."""
    try:
        response = requests.get(
            "https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": "KEY", "units": "metric"}, timeout=10
        )
        if response.status_code == 404:
            return f"Error: City '{city}' not found. Use the English name: 'Mexico City', not 'Ciudad de México'."
        response.raise_for_status()
        data = response.json()
        return f"{city}: {data['main']['temp']}°C, {data['weather'][0]['description']}"
    except requests.Timeout:
        return "Error: The API did not respond. Try again."
    except Exception as e:
        return f"Error: {str(e)}"

If the model passes city: "Ciudad de México" and receives the error "Use the English name: 'Mexico City'", on the next iteration it calls with city="Mexico City". Self-correction with no human intervention.

Common Errors by Type

Each error type has its own pattern. This example consolidates them all into one tool:

import os
import requests
from langchain_core.tools import tool

@tool
def robust_api_call(query: str) -> str:
    """An example handling every common error type."""
    api_key = os.getenv("API_KEY")
    if not api_key:
        return "Error: API_KEY not configured. Answer with what you already know."
    try:
        response = requests.get("https://api.example.com/data",
            params={"q": query, "key": api_key}, timeout=(3, 10))
        if response.status_code == 429:
            return f"Error: Rate limit. Wait {response.headers.get('Retry-After', 'a few seconds')}."
        if response.status_code == 401:
            return "Error: Invalid or expired API key."
        response.raise_for_status()
        return response.json()["result"]
    except requests.ConnectTimeout:
        return "Error: Could not connect (3s timeout)."
    except requests.ReadTimeout:
        return "Error: The server did not respond (10s timeout). The query is too complex."
    except requests.ConnectionError:
        return "Error: No connection. The service may be down."
    except requests.HTTPError as e:
        return f"HTTP error {e.response.status_code}: {e.response.reason}"
    except (KeyError, IndexError, ValueError) as e:
        return f"Error: Unexpected response: {str(e)}"
    except Exception as e:
        return f"Unexpected error: {type(e).__name__}: {str(e)}"

The tuple (3, 10) separates the connect timeout from the read timeout. Checking the API key before the request avoids an HTTP call you know will fail.

Retry Logic

The problem: transient errors

Not every error is permanent. A timeout can be a one-off. A 429 means "try later." A 503 means "I'm temporarily down." For these transient errors, retrying makes sense. For a 404 or invalid input, retrying changes nothing.

Manual retry with exponential backoff

import time
import requests
from langchain_core.tools import tool

def fetch_with_retry(url: str, params: dict, max_retries: int = 3) -> requests.Response:
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=10)
            if response.status_code == 429:
                time.sleep(2 ** attempt)  # 1s, 2s, 4s
                continue
            response.raise_for_status()
            return response
        except (requests.Timeout, requests.ConnectionError):
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
    raise requests.HTTPError("Max retries exceeded")

@tool
def search_with_retry(query: str) -> str:
    """Search for information with automatic retries."""
    try:
        response = fetch_with_retry("https://api.example.com/search", {"q": query})
        return response.json()["results"][0]["snippet"]
    except requests.Timeout:
        return "Error: The search failed after 3 retries due to timeouts."
    except Exception as e:
        return f"Error after retries: {str(e)}"

Retry with tenacity (production)

For production, tenacity simplifies retry logic:

import requests
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, stop_after_delay,
)
from langchain_core.tools import tool

@retry(
    stop=(stop_after_attempt(3) | stop_after_delay(30)),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((requests.Timeout, requests.ConnectionError)),
)
def fetch_weather_api(city: str) -> dict:
    """Call the API with automatic retries for transient errors."""
    response = requests.get(
        "https://api.openweathermap.org/data/2.5/weather",
        params={"q": city, "appid": "KEY", "units": "metric"},
        timeout=10
    )
    response.raise_for_status()
    return response.json()

@tool
def get_weather_robust(city: str) -> str:
    """Get the weather with automatic retries."""
    try:
        data = fetch_weather_api(city)
        return f"{city}: {data['main']['temp']}°C, {data['weather'][0]['description']}"
    except requests.HTTPError as e:
        if e.response.status_code == 404:
            return f"Error: City '{city}' not found."
        return f"Error: The API returned {e.response.status_code} after retries."
    except Exception as e:
        return f"Error: {str(e)}"

Notice the separation: fetch_weather_api handles retries (infrastructure), get_weather_robust handles error presentation (tool logic). stop_after_delay(30) guarantees you never wait more than 30s in total. retry_if_exception_type only retries transient errors — an HTTPError (404, 401) propagates immediately.

What to retry and what not to

Graceful Degradation

Graceful degradation means that when a tool fails, the agent doesn't die — it adapts.

Pattern 1: Fallback tools

import requests
from langchain_core.tools import tool

def _search_tavily(query: str) -> str:
    response = requests.post(
        "https://api.tavily.com/search",
        json={"api_key": "KEY", "query": query, "max_results": 3},
        timeout=10
    )
    response.raise_for_status()
    results = response.json()["results"]
    return "\n".join(f"- {r['title']}: {r['content'][:150]}" for r in results)

def _search_duckduckgo(query: str) -> str:
    response = requests.get(
        "https://api.duckduckgo.com/",
        params={"q": query, "format": "json"},
        timeout=10
    )
    response.raise_for_status()
    return response.json().get("AbstractText", f"No detailed results for '{query}'.")

@tool
def search_web(query: str) -> str:
    """Search the web with high availability."""
    try:
        return _search_tavily(query)
    except Exception as e_primary:
        try:
            result = _search_duckduckgo(query)
            return f"[DuckDuckGo fallback] {result}"
        except Exception as e_fallback:
            return (
                f"Error: Both services failed. "
                f"Answer with the information you already have."
            )

The [DuckDuckGo fallback] prefix tells the model the results come from the secondary service.

Pattern 2: Partial data

When your tool queries multiple sources, return whatever you got instead of failing entirely:

@tool
def get_company_info(company: str) -> str:
    """Get financial information about a company."""
    info = {}
    for field, url in [("price", f"api/price/{company}"), ("news", f"api/news/{company}")]:
        try:
            resp = requests.get(url, timeout=5)
            resp.raise_for_status()
            info[field] = resp.json().get("data", "no data")
        except Exception:
            info[field] = "unavailable"

    available = {k: v for k, v in info.items() if v != "unavailable"}
    if not available:
        return f"Error: Could not fetch information for {company}."
    return f"Info for {company} (partial data possible): {info}"

Pattern 3: Cache as a fallback

import requests
from langchain_core.tools import tool

_cache: dict[str, str] = {}

@tool
def get_weather_cached(city: str) -> str:
    """Get the weather. Falls back to the cache if the API fails."""
    try:
        response = requests.get(
            "https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": "KEY", "units": "metric"},
            timeout=10
        )
        response.raise_for_status()
        data = response.json()
        result = f"{city}: {data['main']['temp']}°C, {data['weather'][0]['description']}"
        _cache[city.lower()] = result
        return result
    except Exception:
        cached = _cache.get(city.lower())
        if cached:
            return f"[Cached data, may not be current] {cached}"
        return f"Error: Could not fetch the weather for {city} and there's no cached data."

Comparison: Error Handling Strategies

The non-negotiable minimum: return an error string in every tool. External APIs: retry + return the error. Critical services: retry + fallback.

Connection to the Project

In this module's project (capsule 08), you'll build an agent with 5 real external tools. Each one must implement error handling:

• Tavily Search: Retry + fallback to DuckDuckGo
• Weather API: City validation + timeout handling
• Calculator: Safe-expression validation with regex
• File Reader: FileNotFoundError, PermissionError, files that are too large
• DateTime: try/except as good practice

The goal: you should be able to unplug your internet, run the agent, and have it keep responding with informative errors — not tracebacks.

In the evolving project (M4-10):

• M3: Retry and fallback get integrated with parallel function calling and circuit breakers
• M4: Error handling inside StateGraph nodes
• M9: "Failure injection" — tests where the tools fail on purpose to verify recovery

Troubleshooting

Problem 1: "A ValidationError crashes my agent"

Cause: The tool invocation isn't wrapped in try/except in the execution loop.

Fix: The handling belongs in the loop:

for tc in response.tool_calls:
    try:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
    except Exception as e:
        result = f"Error invoking {tc['name']}: {e}. Check the arguments."
    messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

Problem 2: "The retry hangs forever"

Cause: No global timeout, or backoff that's too aggressive.

Fix: Bound the total time: stop=(stop_after_attempt(3) | stop_after_delay(30)) in tenacity.

Problem 3: "The model keeps using a tool that always fails"

Cause: The error message doesn't tell it to use an alternative.

Fix: Be explicit:

return (
    "Error: Service unavailable after 3 retries. "
    "Do NOT call this tool again. Answer with what you already have."
)

Problem 4: "Tenacity retries permanent errors"

Cause: An unfiltered retry retries everything, even 404s.

Fix: retry=retry_if_exception_type((requests.Timeout, requests.ConnectionError)) — an HTTPError (404) propagates immediately.

Problem 5: "The error messages confuse the model"

Cause: Technical messages with no actionable context.

Fix: Map errors to useful messages:

ERROR_MESSAGES = {
    requests.Timeout: "The service took too long. Try a shorter query.",
    requests.ConnectionError: "No connection to the service. It may be down.",
    requests.HTTPError: "The service returned an error. Try different parameters.",
}

def friendly_error(e: Exception) -> str:
    for error_type, msg in ERROR_MESSAGES.items():
        if isinstance(e, error_type):
            return f"Error: {msg}"
    return f"Unexpected error: {type(e).__name__}."

Exercises

Exercise 1: A tool with full validation (Easy)

Create convert_temperature(value, from_unit, to_unit) with a Pydantic schema: value between -273.15 and 10000, units limited to "celsius"/"fahrenheit"/"kelvin" (an enum), and from_unit != to_unit (a model_validator). Include error handling.

from langchain_core.tools import tool
from pydantic import BaseModel, Field, model_validator
from enum import Enum

class TempUnit(str, Enum):
    celsius = "celsius"
    fahrenheit = "fahrenheit"
    kelvin = "kelvin"

class ConvertTempInput(BaseModel):
    value: float = Field(ge=-273.15, le=10000, description="The temperature to convert")
    from_unit: TempUnit = Field(description="Source unit")
    to_unit: TempUnit = Field(description="Target unit")

    @model_validator(mode="after")
    def units_must_differ(self):
        if self.from_unit == self.to_unit:
            raise ValueError(f"from_unit and to_unit cannot be the same ({self.from_unit.value})")
        return self

@tool(args_schema=ConvertTempInput)
def convert_temperature(value: float, from_unit: str, to_unit: str) -> str:
    """Convert a temperature between Celsius, Fahrenheit and Kelvin."""
    try:
        if from_unit == "fahrenheit":
            celsius = (value - 32) * 5/9
        elif from_unit == "kelvin":
            celsius = value - 273.15
        else:
            celsius = value

        if to_unit == "fahrenheit":
            result = celsius * 9/5 + 32
        elif to_unit == "kelvin":
            result = celsius + 273.15
        else:
            result = celsius
        return f"{value}° {from_unit} = {round(result, 2)}° {to_unit}"
    except Exception as e:
        return f"Conversion error: {str(e)}"

print(convert_temperature.invoke({"value": 100, "from_unit": "celsius", "to_unit": "fahrenheit"}))
# "100° celsius = 212.0° fahrenheit"

Exercise 2: A tool with manual retries (Medium)

Implement fetch_news(topic: str) that calls an API with manual retries (no tenacity). It retries up to 3 times with waits of 1s, 2s, 4s. It only retries timeouts and 5xx errors. It returns an error if all attempts fail.

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

class NewsInput(BaseModel):
    topic: str = Field(min_length=2, max_length=100, description="The news topic")

@tool(args_schema=NewsInput)
def fetch_news(topic: str) -> str:
    """Search for recent news about a topic."""
    url = "https://newsapi.org/v2/everything"
    params = {"q": topic, "pageSize": 5, "apiKey": "KEY"}
    last_error = None

    for attempt in range(3):
        try:
            response = requests.get(url, params=params, timeout=10)
            if response.status_code >= 500:
                last_error = f"Server error {response.status_code}"
                time.sleep(2 ** attempt)
                continue
            if response.status_code == 401:
                return "Error: Invalid API key. I can't search for news."
            response.raise_for_status()
            articles = response.json().get("articles", [])
            if not articles:
                return f"No news about '{topic}'."
            return "\n".join(f"- {a['title']}" for a in articles[:5])
        except requests.Timeout:
            last_error = "timeout"
            if attempt < 2:
                time.sleep(2 ** attempt)
        except requests.ConnectionError:
            return "Error: No connection to the news service."
        except Exception as e:
            return f"Unexpected error: {str(e)}"

    return f"Error: News lookup failed after 3 attempts. Last error: {last_error}."

Exercise 3: A 3-level fallback chain (Medium)

Create translate_text(text, target_language) that tries: (1) a primary API, (2) a secondary API, (3) returning the original text with a message that it couldn't be translated.

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

class TranslateInput(BaseModel):
    text: str = Field(min_length=1, max_length=5000, description="The text to translate")
    target_language: str = Field(min_length=2, max_length=5, description="Code: 'es', 'en', 'fr'")

def _translate_primary(text: str, target: str) -> str:
    resp = requests.post("https://api.deepl.com/v2/translate",
        data={"text": text, "target_lang": target.upper(), "auth_key": "KEY"}, timeout=10)
    resp.raise_for_status()
    return resp.json()["translations"][0]["text"]

def _translate_secondary(text: str, target: str) -> str:
    resp = requests.post("https://libretranslate.com/translate",
        json={"q": text, "source": "auto", "target": target}, timeout=15)
    resp.raise_for_status()
    return resp.json()["translatedText"]

@tool(args_schema=TranslateInput)
def translate_text(text: str, target_language: str) -> str:
    """Translate text into another language with high availability."""
    errors = []
    try:
        return f"[Translation] {_translate_primary(text, target_language)}"
    except Exception as e:
        errors.append(f"DeepL: {type(e).__name__}")
    try:
        return f"[Alternative] {_translate_secondary(text, target_language)}"
    except Exception as e:
        errors.append(f"LibreTranslate: {type(e).__name__}")
    return f"Error: Could not translate ({', '.join(errors)}). Text: '{text[:100]}...'"

Exercise 4: Error handling in the execution loop (Medium)

Modify capsule 04's agent_loop to: (1) catch ValidationError with a descriptive message, (2) handle a tool that isn't found, (3) apply a 60s global timeout, (4) count errors per tool.

import time
from pydantic import ValidationError
from langchain_core.messages import HumanMessage, ToolMessage

def agent_loop_robust(user_input, model_with_tools, tools_by_name, max_iterations=5):
    messages = [HumanMessage(content=user_input)]
    error_counts = {}
    start_time = time.time()

    for i in range(max_iterations):
        if time.time() - start_time > 60.0:
            return {"response": "Global timeout (60s).", "error_counts": error_counts}

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

        if not response.tool_calls:
            return {"response": response.content or "No answer", "error_counts": error_counts}

        for tc in response.tool_calls:
            name = tc["name"]
            if name not in tools_by_name:
                result = f"Error: Tool '{name}' does not exist. Available: {list(tools_by_name.keys())}"
                error_counts[name] = error_counts.get(name, 0) + 1
            else:
                try:
                    result = str(tools_by_name[name].invoke(tc["args"]))
                except ValidationError as e:
                    error_counts[name] = error_counts.get(name, 0) + 1
                    errors = "; ".join(f"{err['loc']}: {err['msg']}" for err in e.errors())
                    result = f"Validation error in {name}: {errors}. Check the arguments."
                except Exception as e:
                    error_counts[name] = error_counts.get(name, 0) + 1
                    result = f"Error running {name}: {type(e).__name__}: {str(e)}"

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

    return {"response": "Iteration limit.", "error_counts": error_counts}

Exercise 5: A complete production-ready tool (Hard)

Create search_github_repos(query, language, min_stars) with: (1) a Pydantic schema (query min 2 chars, language enum, min_stars >= 0), (2) retries with tenacity for timeouts, (3) a fallback with hardcoded repos if GitHub fails, (4) actionable messages. Use https://api.github.com/search/repositories.

import requests
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from enum import Enum
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class Language(str, Enum):
    python = "python"
    javascript = "javascript"
    typescript = "typescript"
    rust = "rust"
    any = "any"

class GitHubSearchInput(BaseModel):
    query: str = Field(min_length=2, max_length=200, description="Repo search term")
    language: Language = Field(default=Language.any, description="Filter by language")
    min_stars: int = Field(default=0, ge=0, le=500000, description="Minimum stars")

FALLBACK = {"python": "langchain, fastapi, pydantic", "any": "linux, vscode, tensorflow"}

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=8),
    retry=retry_if_exception_type((requests.Timeout, requests.ConnectionError)),
)
def _github_search(query: str, language: str, min_stars: int) -> list[dict]:
    q = query + (f" language:{language}" if language != "any" else "")
    q += f" stars:>={min_stars}" if min_stars > 0 else ""
    resp = requests.get("https://api.github.com/search/repositories",
        params={"q": q, "sort": "stars", "per_page": 5}, timeout=10)
    resp.raise_for_status()
    return resp.json().get("items", [])

@tool(args_schema=GitHubSearchInput)
def search_github_repos(query: str, language: str = "any", min_stars: int = 0) -> str:
    """Search GitHub repos by topic and popularity.
    Use for: libraries, frameworks, open source projects.
    Do NOT use for: finding people, issues, or code inside files."""
    try:
        items = _github_search(query, language, min_stars)
        if not items:
            return f"No repos for '{query}' with those filters."
        lines = [f"- {r['full_name']} | ⭐ {r['stargazers_count']:,} | {r.get('description', '')[:80]}"
                 for r in items]
        return f"Repos for '{query}':\n" + "\n".join(lines)
    except requests.HTTPError as e:
        if hasattr(e, 'response') and e.response and e.response.status_code == 403:
            return f"[Rate limit] Popular repos: {FALLBACK.get(language, FALLBACK['any'])}"
        return f"HTTP error: {str(e)}"
    except Exception as e:
        return f"Error ({type(e).__name__}). Popular: {FALLBACK.get(language, FALLBACK['any'])}"

Summary

In this capsule you learned:

• Error handling is a first-class concern: A tool without error handling is a bug waiting to happen. It isn't optional
• Never propagate exceptions: Return strings with clear errors as ToolMessages so the model can self-correct
• Pydantic validates before execution: Field(ge=, le=, min_length=, pattern=), field_validator for custom logic, model_validator for cross-field validation
• Each error type has its pattern: Timeout → retry, 429 → backoff, 404 → immediate error, ConnectionError → retry or fallback
• Retry with exponential backoff: Manual (2 ** attempt) or tenacity for production. Only retry transient errors
• Graceful degradation: Fallback tools, partial data, cache as a last resort
• Actionable error messages: What failed + why + what to do
• Separate infrastructure from presentation: The internal function → retries/network. The tool → error presentation

Next capsule: Advanced Tool Schemas — nested Pydantic models, complex enums, Optional fields, and InjectedToolArg to inject dependencies (user_id, session, DB connection) without the LLM seeing them as arguments.

Additional Resources

1. Pydantic Validators — The complete reference for field_validator and model_validator
2. Tenacity Documentation — A retry library for Python, ideal for production
3. LangChain — How to handle tool errors — The official guide to error handling in tools
4. Requests — Timeouts — Connect and read timeouts
5. Circuit Breaker Pattern — A production pattern for unstable services
6. Exponential Backoff and Jitter — Retry strategies in distributed systems