8. Project: Agent with 5 Real External Tools

Project overview

Throughout this module you learned to create tools with @tool, to design Pydantic schemas that guide the model, to implement the tool execution loop step by step, to connect tools to real external APIs, and to handle errors without crashing the agent. Now you'll pull it all together into a concrete project: build an agent with 5 tools that connect to real APIs and real functionality — no mocks, no hardcoded data, no simulations.

Why "real"? Because a tool that returns "mock result" teaches you nothing that matters in production. You don't see real latency, real rate limits, real timeouts. When your weather tool calls Open-Meteo and the API takes 800ms to respond, you learn something no mock can teach you: the agent needs to handle that latency without breaking. When web search returns empty results for an odd query, you learn that your tool needs a coherent fallback.

The 5 tools cover the full spectrum of what a real agent needs: searching for information (web search), querying real-time data (weather), processing data (calculator), accessing temporal context (date and time), and reading local files (file reader). Each one has its own error type, its own Pydantic schema with real validation, and its own granular error handling.

The result is an agent you can ask "Look up the temperature in Berlin, convert it to Fahrenheit, and tell me what day it is" — and the agent orchestrates three tools, combines the results, and gives you a coherent answer. If one tool fails, the rest keep working.

Estimated time: 45-60 minutes.

---

Project Goal

Build a working agent with 5 real external tools, each with a complete Pydantic schema, robust error handling, and input validation.

By the end you'll be able to:

• Create tools that connect to real external APIs with requests and specialized packages
• Design Pydantic schemas with constraints (min_length, max_length, ge, le, pattern, field_validator) that prevent invalid inputs
• Implement granular error handling by failure type: network timeouts, API errors, invalid inputs, files not found
• Combine 5 tools into an agent with create_react_agent that resolves multi-tool queries
• Verify that one tool failing neither crashes the agent nor blocks the others

---

Technical Specifications

Stack

Technology	Version	Use
Python	3.11+	Runtime
langchain	v1.2+	@tool, init_chat_model
langgraph	v1.0+	create_react_agent
langchain-openai	latest	OpenAI provider
requests	latest	HTTP requests to APIs
duckduckgo-search	latest	Web search without an API key
python-dotenv	any	Environment variables

Setup

pip install langchain langchain-openai langgraph requests duckduckgo-search python-dotenv

Create a .env file:

OPENAI_API_KEY=sk-proj-your-api-key-here

You only need the OpenAI key. The other tools (DuckDuckGo, Open-Meteo, calculator, datetime, file reader) require no key.

Project files

agent_5_tools/
├── .env                  # OpenAI API key
├── complete_agent.py     # The whole project in one file
└── sample_files/
    ├── notes.txt         # Test file
    └── data.csv          # Test CSV file

Before you start, create the test files for the read_file tool:

mkdir -p agent_5_tools/sample_files

sample_files/notes.txt:

Team meeting - March 8, 2026
====================================
Attendees: Ana, Carlos, Maria, Pedro
Topic: Sprint 14 review

Decisions:
1. Migrate the database to PostgreSQL before March 15
2. Implement OAuth2 authentication in the API gateway
3. Review the search service's performance (P95 > 500ms)

Next meeting: March 15, 2026, 10:00 AM

sample_files/data.csv:

product,price,stock
Laptop Pro,1299.99,45
Monitor 4K,549.99,120
Mechanical Keyboard,89.99,200
Ergonomic Mouse,59.99,150
HD Webcam,79.99,80

---

The 5 Tools — Specification

Before writing any code, here's the specification for each tool.

#	Tool	API / Source	Key	Main error type
1	web_search	DuckDuckGo	No	Rate limit, empty results
2	get_weather	Open-Meteo	No	City not found, timeout
3	calculator	ast module	No	Invalid expression, div/0
4	get_datetime	datetime + zoneinfo	No	Invalid timezone
5	read_file	Local filesystem	No	File not found, path traversal

Every tool follows the same pattern:

1. Pydantic schema → validates inputs before execution
2. @tool function → implementation with complete error handling
3. Returns a string → on both success and error, so the model always gets feedback

---

Step-by-Step Implementation

Step 1: Imports and base configuration

import os
import re
import ast
import requests
from datetime import datetime
from typing import Optional
from zoneinfo import ZoneInfo, available_timezones

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field, field_validator
from duckduckgo_search import DDGS

Step 2: Tool 1 — Web Search (DuckDuckGo)

Web search without an API key. The duckduckgo-search package makes requests directly to the DuckDuckGo engine.

class WebSearchInput(BaseModel):
    query: str = Field(
        min_length=2,
        max_length=200,
        description="Search term. Be specific for better results."
    )
    max_results: int = Field(
        default=3,
        ge=1,
        le=10,
        description="Maximum number of results to return (1-10)."
    )

@tool(args_schema=WebSearchInput)
def web_search(query: str, max_results: int = 3) -> str:
    """Search the web using DuckDuckGo. Use for questions about
    current events, specific facts, or any information that
    requires an internet search."""
    try:
        with DDGS() as ddgs:
            results = list(ddgs.text(query, max_results=max_results))

        if not results:
            return f"No results found for: '{query}'"

        formatted = []
        for i, r in enumerate(results, 1):
            title = r.get("title", "No title")
            body = r.get("body", "No description")
            url = r.get("href", "")
            formatted.append(f"{i}. **{title}**\n   {body}\n   URL: {url}")

        return f"Results for '{query}':\n\n" + "\n\n".join(formatted)

    except Exception as e:
        return f"Web search error: {str(e)}. Try rephrasing the query."

Two key validations in the schema: min_length=2 prevents empty searches that return garbage, and max_results with ge=1, le=10 stops the model from asking for 0 results (useless) or 100 (excessive).

Direct test:

result = web_search.invoke({"query": "LangGraph create_react_agent 2026", "max_results": 2})
print(result)
# Results for 'LangGraph create_react_agent 2026':
# 1. **Build an Agent with LangGraph - LangChain**
#    Learn how to build an agent using LangGraph's create_react_agent...
#    URL: https://python.langchain.com/docs/tutorials/agents/

Step 3: Tool 2 — Weather (Open-Meteo API)

Real-time weather using Open-Meteo — completely free, no key. It requires two HTTP requests: geocoding (name → coordinates) and forecast (coordinates → weather).

class WeatherInput(BaseModel):
    city: str = Field(
        min_length=1,
        max_length=100,
        description="City name: 'Madrid', 'Mexico City', 'Tokyo'."
    )
    units: str = Field(
        default="celsius",
        pattern="^(celsius|fahrenheit)$",
        description="Temperature units: 'celsius' or 'fahrenheit'."
    )

WEATHER_CONDITIONS = {
    0: "clear", 1: "mostly clear", 2: "partly cloudy",
    3: "cloudy", 45: "fog", 48: "freezing fog",
    51: "light drizzle", 53: "moderate drizzle", 55: "heavy drizzle",
    61: "light rain", 63: "moderate rain", 65: "heavy rain",
    71: "light snow", 73: "moderate snow", 75: "heavy snow",
    80: "light showers", 81: "moderate showers", 82: "heavy showers",
    95: "thunderstorm", 96: "thunderstorm with light hail",
    99: "thunderstorm with heavy hail",
}

@tool(args_schema=WeatherInput)
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city using real-time data.
    Reports temperature, condition, humidity and wind."""
    try:
        geo_response = requests.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params={"name": city, "count": 1, "language": "en"},
            timeout=5,
        )
        geo_response.raise_for_status()
        geo_data = geo_response.json()

        if not geo_data.get("results"):
            return f"City not found: '{city}'. Check the name and try again."

        location = geo_data["results"][0]
        lat = location["latitude"]
        lon = location["longitude"]
        city_name = location.get("name", city)
        country = location.get("country", "")

        temp_unit = "fahrenheit" if units == "fahrenheit" else "celsius"
        weather_response = requests.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": lat,
                "longitude": lon,
                "current": "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code",
                "temperature_unit": temp_unit,
                "wind_speed_unit": "kmh",
            },
            timeout=5,
        )
        weather_response.raise_for_status()
        current = weather_response.json()["current"]

        temp = current["temperature_2m"]
        humidity = current["relative_humidity_2m"]
        wind = current["wind_speed_10m"]
        condition = WEATHER_CONDITIONS.get(current["weather_code"], f"code {current['weather_code']}")
        symbol = "°F" if units == "fahrenheit" else "°C"

        return (
            f"Weather in {city_name}, {country}:\n"
            f"  Temperature: {temp}{symbol}\n"
            f"  Condition: {condition}\n"
            f"  Humidity: {humidity}%\n"
            f"  Wind: {wind} km/h"
        )

    except requests.Timeout:
        return f"Error: timeout while fetching weather for '{city}'. Try again."
    except requests.ConnectionError:
        return f"Error: no internet connection to fetch the weather."
    except requests.HTTPError as e:
        return f"HTTP error from the weather service: {e}"
    except KeyError as e:
        return f"Error: unexpected response from the weather service. Missing field: {e}"
    except Exception as e:
        return f"Unexpected error while fetching weather: {str(e)}"

Five distinct except blocks. That's not paranoia — it's production. Each error type produces a different message that gives the model information about what went wrong. The pattern="^(celsius|fahrenheit)$" in the schema prevents the model from passing "kelvin" or "centigrade".

Direct test:

result = get_weather.invoke({"city": "Madrid"})
print(result)
# Weather in Madrid, Spain:
#   Temperature: 18.3°C
#   Condition: partly cloudy
#   Humidity: 52%
#   Wind: 12.4 km/h

Step 4: Tool 3 — Calculator (Safe Evaluation with AST)

A calculator that uses ast to validate expressions before evaluating them. It doesn't use eval() directly — it parses the AST and verifies it only contains allowed math operations.

class CalculatorInput(BaseModel):
    expression: str = Field(
        min_length=1,
        max_length=200,
        description=(
            "Math expression. Supports: +, -, *, /, ** (power), "
            "// (floor division), % (modulo), and parentheses. "
            "Examples: '15 * 23', '(100 + 50) / 3', '2 ** 10'."
        )
    )

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

ALLOWED_AST_NODES = (
    ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant,
    ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow,
    ast.Mod, ast.FloorDiv, ast.USub, ast.UAdd,
)

@tool(args_schema=CalculatorInput)
def calculator(expression: str) -> str:
    """Evaluate math expressions. Supports +, -, *, /, ** (power),
    // (floor division), % (modulo) and parentheses."""
    try:
        tree = ast.parse(expression, mode="eval")
        for node in ast.walk(tree):
            if not isinstance(node, ALLOWED_AST_NODES):
                return f"Error: operation not allowed in '{expression}'."

        result = eval(compile(tree, "<expr>", "eval"))

        if isinstance(result, float):
            if result == float("inf") or result == float("-inf"):
                return f"Error: infinite result in '{expression}'."
            if result != result:
                return f"Error: non-numeric result in '{expression}'."
            if result == int(result):
                result = int(result)

        return f"{expression} = {result}"

    except ZeroDivisionError:
        return f"Error: division by zero in '{expression}'."
    except SyntaxError:
        return f"Error: invalid syntax in '{expression}'. Check parentheses and operators."
    except Exception as e:
        return f"Error evaluating '{expression}': {str(e)}"

The double validation (Pydantic regex + AST whitelist) is intentional. Pydantic filters dangerous characters at the string level. The AST whitelist filters dangerous nodes at the structure level. If the model passes "__import__('os').system('rm -rf /')", Pydantic rejects it for the disallowed characters before the function ever runs.

Direct test:

print(calculator.invoke({"expression": "1547 * 23 + 890"}))   # 1547 * 23 + 890 = 36471
print(calculator.invoke({"expression": "(100 + 50) / 3"}))     # (100 + 50) / 3 = 50.0
print(calculator.invoke({"expression": "2 ** 32"}))             # 2 ** 32 = 4294967296
print(calculator.invoke({"expression": "10 / 0"}))              # Error: division by zero in '10 / 0'.

Step 5: Tool 4 — Date/Time with Timezone

Current date and time with optional timezone support. Uses zoneinfo (included in Python 3.9+).

class DateTimeInput(BaseModel):
    timezone: Optional[str] = Field(
        default=None,
        max_length=50,
        description=(
            "Optional timezone. Examples: 'America/Mexico_City', 'Europe/Madrid', "
            "'Asia/Tokyo', 'US/Eastern'. If not specified, uses local time."
        )
    )

@tool(args_schema=DateTimeInput)
def get_datetime(timezone: Optional[str] = None) -> str:
    """Get the current date and time. Optionally accepts a timezone.
    Use for questions about date, time, day of the week, or temporal information."""
    try:
        if timezone:
            if timezone not in available_timezones():
                examples = "America/Mexico_City, Europe/Madrid, Asia/Tokyo, US/Eastern, US/Pacific"
                return f"Error: timezone '{timezone}' not recognized. Valid examples: {examples}"
            now = datetime.now(ZoneInfo(timezone))
            location = f" ({timezone})"
        else:
            now = datetime.now()
            location = " (local time)"

        day_name = now.strftime("%A")
        month_name = now.strftime("%B")

        return (
            f"Current date and time{location}:\n"
            f"  Date: {day_name}, {month_name} {now.day}, {now.year}\n"
            f"  Time: {now.strftime('%H:%M:%S')}\n"
            f"  ISO 8601: {now.isoformat()}"
        )

    except Exception as e:
        return f"Error getting date/time: {str(e)}"

timezone is Optional with a None default. When the user asks "what time is it?", the model calls it with no arguments. When they ask "what time is it in Tokyo?", it passes timezone: "Asia/Tokyo". strftime("%A") and strftime("%B") already give you the weekday and month names in English — if you were serving another language, you'd add a translation dictionary here so the agent doesn't answer with a mix of two languages.

Direct test:

print(get_datetime.invoke({}))
# Current date and time (local time):
#   Date: Saturday, March 8, 2026
#   Time: 14:23:45
#   ISO 8601: 2026-03-08T14:23:45.123456

print(get_datetime.invoke({"timezone": "Asia/Tokyo"}))
# Current date and time (Asia/Tokyo):
#   Date: Sunday, March 9, 2026
#   Time: 04:23:45
#   ISO 8601: 2026-03-09T04:23:45.123456+09:00

print(get_datetime.invoke({"timezone": "made_up_zone"}))
# Error: timezone 'made_up_zone' not recognized. Valid examples: ...

Step 6: Tool 5 — File Reader with Sandbox

Reads local text files, restricted to a sandbox directory for safety.

class FileReaderInput(BaseModel):
    file_path: str = Field(
        min_length=1,
        max_length=200,
        description=(
            "File path relative to sample_files/. "
            "Examples: 'notes.txt', 'data.csv'. "
            "Can only read files inside sample_files/."
        )
    )
    max_chars: int = Field(
        default=2000,
        ge=100,
        le=10000,
        description="Maximum characters to read (100-10000). Default: 2000."
    )

SANDBOX_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sample_files")

@tool(args_schema=FileReaderInput)
def read_file(file_path: str, max_chars: int = 2000) -> str:
    """Read the contents of a local text file. Only accesses files
    inside sample_files/. Use to read notes, CSV data, or configs."""
    try:
        safe_path = os.path.normpath(os.path.join(SANDBOX_DIR, file_path))
        if not safe_path.startswith(SANDBOX_DIR):
            return (
                f"Security error: access to '{file_path}' is not allowed. "
                "Only files inside sample_files/."
            )

        if not os.path.exists(safe_path):
            available = os.listdir(SANDBOX_DIR) if os.path.exists(SANDBOX_DIR) else []
            files_list = ", ".join(available) if available else "none"
            return f"Error: '{file_path}' not found. Available files: {files_list}"

        if not os.path.isfile(safe_path):
            return f"Error: '{file_path}' is not a file."

        file_size = os.path.getsize(safe_path)
        with open(safe_path, "r", encoding="utf-8") as f:
            content = f.read(max_chars)

        truncated = " (truncated)" if file_size > max_chars else ""
        return f"File: {file_path} ({file_size} bytes{truncated})\n---\n{content}"

    except PermissionError:
        return f"Error: no permission to read '{file_path}'."
    except UnicodeDecodeError:
        return f"Error: '{file_path}' is not valid text (incompatible encoding)."
    except Exception as e:
        return f"Error reading '{file_path}': {str(e)}"

The safe_path.startswith(SANDBOX_DIR) check is the security barrier. If the model passes "../../etc/passwd", os.path.normpath resolves it to a path outside the sandbox, and the function rejects it. max_chars with ge=100, le=10000 prevents reading fewer than 100 characters (useless) or more than 10,000 (which could flood the model's context).

Direct test:

print(read_file.invoke({"file_path": "notes.txt"}))
# File: notes.txt (389 bytes)
# ---
# Team meeting - March 8, 2026 ...

print(read_file.invoke({"file_path": "../../etc/passwd"}))
# Security error: access to '../../etc/passwd' is not allowed. ...

print(read_file.invoke({"file_path": "does_not_exist.txt"}))
# Error: 'does_not_exist.txt' not found. Available files: notes.txt, data.csv

Step 7: Connect the 5 tools to the agent

ALL_TOOLS = [web_search, get_weather, calculator, get_datetime, read_file]

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(
    model,
    ALL_TOOLS,
    prompt=(
        "You are an assistant with real tools: web search, weather, calculator, "
        "date/time and a file reader. Combine results from multiple tools "
        "into coherent answers. If a tool fails, report what happened. "
        "Always answer in English."
    ),
)

Three design decisions:

1. prompt= in create_react_agent. The system prompt tells the agent to answer in English and to handle errors by informing the user. Without this prompt, the agent might answer in another language or silently ignore errors.

2. All the tools in one list. ALL_TOOLS groups the 5 tools. create_react_agent registers them by name and generates the combined schema the model receives.

3. init_chat_model("openai:gpt-4.1-mini"). A fast, cheap model that handles tool calling correctly. You can switch to "openai:gpt-4.1" for better reasoning on complex queries.

---

The Complete Agent

Here's the whole project consolidated into a single runnable file. Copy this code into complete_agent.py and run it.

# complete_agent.py
import os
import re
import ast
import requests
from datetime import datetime
from typing import Optional
from zoneinfo import ZoneInfo, available_timezones

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field, field_validator
from duckduckgo_search import DDGS


# ============================================================
# Pydantic Schemas
# ============================================================

class WebSearchInput(BaseModel):
    query: str = Field(min_length=2, max_length=200, description="Search term.")
    max_results: int = Field(default=3, ge=1, le=10, description="Results (1-10).")

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

class CalculatorInput(BaseModel):
    expression: str = Field(min_length=1, max_length=200, description="Math expression.")

    @field_validator("expression")
    @classmethod
    def validate_safe(cls, v: str) -> str:
        if not re.match(r'^[\d\s\+\-\*\/\(\)\.\%]+$', v):
            raise ValueError(f"Characters not allowed: '{v}'")
        return v.strip()

class DateTimeInput(BaseModel):
    timezone: Optional[str] = Field(default=None, max_length=50, description="Timezone.")

class FileReaderInput(BaseModel):
    file_path: str = Field(min_length=1, max_length=200, description="Path inside sample_files/.")
    max_chars: int = Field(default=2000, ge=100, le=10000, description="Max characters.")


# ============================================================
# Tools
# ============================================================

@tool(args_schema=WebSearchInput)
def web_search(query: str, max_results: int = 3) -> str:
    """Search the web with DuckDuckGo."""
    try:
        with DDGS() as ddgs:
            results = list(ddgs.text(query, max_results=max_results))
        if not results:
            return f"No results for: '{query}'"
        formatted = []
        for i, r in enumerate(results, 1):
            formatted.append(
                f"{i}. {r.get('title', 'N/A')}\n   {r.get('body', 'N/A')}\n   {r.get('href', '')}"
            )
        return f"Results for '{query}':\n\n" + "\n\n".join(formatted)
    except Exception as e:
        return f"Search error: {e}"


WEATHER_CONDITIONS = {
    0: "clear", 1: "mostly clear", 2: "partly cloudy", 3: "cloudy",
    45: "fog", 51: "drizzle", 61: "light rain", 63: "moderate rain",
    65: "heavy rain", 71: "light snow", 80: "showers", 95: "thunderstorm",
}

@tool(args_schema=WeatherInput)
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city with real-time data."""
    try:
        geo = requests.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params={"name": city, "count": 1, "language": "en"}, timeout=5,
        )
        geo.raise_for_status()
        if not geo.json().get("results"):
            return f"City not found: '{city}'"
        loc = geo.json()["results"][0]

        weather = requests.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": loc["latitude"], "longitude": loc["longitude"],
                "current": "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code",
                "temperature_unit": "fahrenheit" if units == "fahrenheit" else "celsius",
                "wind_speed_unit": "kmh",
            }, timeout=5,
        )
        weather.raise_for_status()
        c = weather.json()["current"]
        sym = "°F" if units == "fahrenheit" else "°C"
        cond = WEATHER_CONDITIONS.get(c["weather_code"], f"code {c['weather_code']}")
        return (
            f"Weather in {loc.get('name', city)}, {loc.get('country', '')}:\n"
            f"  {c['temperature_2m']}{sym}, {cond}, humidity {c['relative_humidity_2m']}%, "
            f"wind {c['wind_speed_10m']} km/h"
        )
    except requests.Timeout:
        return f"Error: timeout fetching weather for '{city}'."
    except requests.ConnectionError:
        return "Error: no internet connection."
    except Exception as e:
        return f"Weather error: {e}"


ALLOWED_NODES = (
    ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant,
    ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow,
    ast.Mod, ast.FloorDiv, ast.USub, ast.UAdd,
)

@tool(args_schema=CalculatorInput)
def calculator(expression: str) -> str:
    """Evaluate math expressions: +, -, *, /, **, //, %, parentheses."""
    try:
        tree = ast.parse(expression, mode="eval")
        for node in ast.walk(tree):
            if not isinstance(node, ALLOWED_NODES):
                return f"Error: operation not allowed in '{expression}'."
        result = eval(compile(tree, "<expr>", "eval"))
        if isinstance(result, float) and result == int(result):
            result = int(result)
        return f"{expression} = {result}"
    except ZeroDivisionError:
        return f"Error: division by zero in '{expression}'."
    except SyntaxError:
        return f"Error: invalid syntax in '{expression}'."
    except Exception as e:
        return f"Error: {e}"


@tool(args_schema=DateTimeInput)
def get_datetime(timezone: Optional[str] = None) -> str:
    """Get the current date and time, with an optional timezone."""
    try:
        if timezone:
            if timezone not in available_timezones():
                return (
                    f"Timezone '{timezone}' not recognized. "
                    "Use: America/Mexico_City, Europe/Madrid, Asia/Tokyo"
                )
            now = datetime.now(ZoneInfo(timezone))
            loc = f" ({timezone})"
        else:
            now = datetime.now()
            loc = " (local time)"
        day = now.strftime("%A")
        month = now.strftime("%B")
        return (
            f"Date and time{loc}: {day}, {month} {now.day}, {now.year}, "
            f"{now.strftime('%H:%M:%S')}"
        )
    except Exception as e:
        return f"Error: {e}"


SANDBOX = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sample_files")

@tool(args_schema=FileReaderInput)
def read_file(file_path: str, max_chars: int = 2000) -> str:
    """Read text files in sample_files/. For notes, CSVs, configs."""
    try:
        safe = os.path.normpath(os.path.join(SANDBOX, file_path))
        if not safe.startswith(SANDBOX):
            return f"Security error: access denied to '{file_path}'."
        if not os.path.exists(safe):
            avail = os.listdir(SANDBOX) if os.path.exists(SANDBOX) else []
            return f"'{file_path}' not found. Available: {', '.join(avail) or 'none'}"
        if not os.path.isfile(safe):
            return f"Error: '{file_path}' is not a file."
        size = os.path.getsize(safe)
        with open(safe, "r", encoding="utf-8") as f:
            content = f.read(max_chars)
        trunc = " (truncated)" if size > max_chars else ""
        return f"File: {file_path} ({size} bytes{trunc})\n---\n{content}"
    except PermissionError:
        return f"Error: no permission to read '{file_path}'."
    except UnicodeDecodeError:
        return f"Error: '{file_path}' is not valid text."
    except Exception as e:
        return f"Error: {e}"


# ============================================================
# Agent
# ============================================================

ALL_TOOLS = [web_search, get_weather, calculator, get_datetime, read_file]

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(
    model,
    ALL_TOOLS,
    prompt=(
        "You are an assistant with real tools: web search, weather, calculator, "
        "date/time and a file reader. Combine results from multiple tools "
        "into coherent answers. If a tool fails, report what happened. "
        "Always answer in English."
    ),
)


def run_query(query: str) -> None:
    """Run a query and show the full flow."""
    print(f"\n{'='*60}")
    print(f"  Query: {query}")
    print(f"{'='*60}")

    result = agent.invoke({"messages": [("user", query)]})

    for msg in result["messages"]:
        name = type(msg).__name__
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            tools = [tc["name"] for tc in msg.tool_calls]
            print(f"  {name}: [tools: {tools}]")
        elif name == "ToolMessage":
            preview = msg.content[:100] + "..." if len(msg.content) > 100 else msg.content
            print(f"  {name}: {preview}")

    final = result["messages"][-1].content
    print(f"\n  Answer: {final}")


if __name__ == "__main__":
    run_query("What's the weather in Madrid and what day is it today?")
    run_query("Search for information about LangGraph and calculate 2**16")
    run_query("Read the notes.txt file and tell me what time it is in Tokyo")
    run_query("What is (1500 * 12 + 500 * 6) / 24?")
    run_query("Look up the current price of Bitcoin")

---

Recommended Tests

Run these queries to validate every scenario.

Single tool

run_query("What is Model Context Protocol and who created it?")

Expected: Only web_search. Returns information about MCP/Anthropic.

run_query("What is 1547 * 23 + 890?")

Expected: Only calculator. Returns 36471.

Multi-tool

run_query("What's the weather in Mexico City and what time is it there?")

Expected: get_weather + get_datetime with timezone America/Mexico_City. The model generates both tool_calls in parallel.

Expected output:

============================================================
  Query: What's the weather in Mexico City and what time is it there?
============================================================
  AIMessage: [tools: ['get_weather', 'get_datetime']]
  ToolMessage: Weather in Mexico City, Mexico:
    18.5°C, partly cloudy, humidity 58%, wind 8.3 km/h
  ToolMessage: Date and time (America/Mexico_City): Saturday, March 8, 2026, 12:23:45

  Answer: The weather in Mexico City is 18.5°C, partly cloudy
  with 58% humidity and wind at 8.3 km/h. It's 12:23 there on Saturday,
  March 8, 2026.

run_query("Read the notes.txt file and tell me what day it is today so I know whether the next meeting already happened")

Expected: read_file + get_datetime. The agent reads the file, gets the current date, and compares them to give a contextual answer.

Triple tool

run_query("What's the temperature in Berlin in Fahrenheit, what is that squared, and what's today's date?")

Expected: get_weather(units="fahrenheit") + calculator + get_datetime. The agent combines all three results.

Error scenarios

run_query("What's the weather in Xyzzyville?")

Expected: get_weather returns "City not found: 'Xyzzyville'". The agent informs the user without crashing.

run_query("Read the secret.txt file")

Expected: read_file returns an error listing the available files. The agent can suggest existing files.

run_query("What is 100 / 0?")

Expected: calculator returns "Error: division by zero". The agent explains that the operation isn't possible.

No tools

run_query("What is the capital of Japan?")

Expected: Answers directly without calling any tool. Zero tool calls.

---

Success Criteria

1. All 5 tools work with real APIs/functionality

No mocks. web_search returns results from real DuckDuckGo. get_weather returns a real temperature from Open-Meteo. calculator evaluates expressions with AST. get_datetime returns the real date/time. read_file reads real files from the filesystem.

2. An error in one tool doesn't crash the agent

When get_weather gets a nonexistent city, it returns an error string — it doesn't raise an exception. The agent keeps working and can use other tools or inform the user.

3. Pydantic schemas validate inputs correctly

Every tool has a schema with constraints. If the model passes invalid inputs (empty string, unsupported units, an expression with dangerous characters), Pydantic rejects them before execution.

4. The agent combines results from multiple tools

Multi-tool queries produce an answer that integrates every result coherently, not three separate answers.

---

Completion Checklist

Setup:

• .env with OPENAI_API_KEY configured
• Dependencies installed: langchain, langgraph, langchain-openai, requests, duckduckgo-search, python-dotenv
• sample_files/ directory with notes.txt and data.csv

Tools:

• web_search returns real DuckDuckGo results
• get_weather returns real weather from Open-Meteo
• calculator evaluates expressions with AST validation
• get_datetime returns date/time with an optional timezone
• read_file reads files with a security sandbox

Pydantic Schemas:

• WebSearchInput: min_length, max_length, ge, le
• WeatherInput: min_length, max_length, pattern
• CalculatorInput: field_validator with regex + AST whitelist
• DateTimeInput: Optional field with max_length
• FileReaderInput: min_length, max_length, ge, le

Error Handling:

• web_search: catches exceptions from the duckduckgo-search package
• get_weather: catches Timeout, ConnectionError, HTTPError, KeyError
• calculator: catches ZeroDivisionError, SyntaxError, disallowed AST nodes
• get_datetime: validates the timezone against available_timezones()
• read_file: validates path traversal, nonexistent file, permissions, encoding

Agent:

• Uses create_react_agent from langgraph.prebuilt (not legacy)
• System prompt in English
• Resolves single-tool queries
• Resolves multi-tool queries (2-3 tools at once)
• Handles tool errors without crashing
• Answers directly when it doesn't need tools

---

Common Errors

Error 1: DuckDuckGo raises RatelimitException

Symptom: duckduckgo_search.exceptions.RatelimitException after many searches in a row.

Cause: DuckDuckGo has implicit rate limiting. More than ~15-20 queries in a minute gets you temporarily blocked.

Solution: Catch the specific exception:

from duckduckgo_search.exceptions import RatelimitException

try:
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=max_results))
except RatelimitException:
    return "Error: search limit reached. Wait a few seconds."

Error 2: Open-Meteo returns the wrong city for ambiguous names

Symptom: You ask for "San José" and get the weather in Costa Rica when you wanted California.

Solution: Include the country in the query: "San José, Costa Rica" or "San José, California". The schema description should guide the model: "For ambiguous names, include the country.".

Error 3: Importing create_react_agent from the wrong place

Symptom: ImportError, or the agent requires an AgentExecutor wrapper.

Cause: There are two versions:

# ❌ LEGACY: needs AgentExecutor
from langchain.agents import create_react_agent, AgentExecutor

# ✅ MODERN: works directly
from langgraph.prebuilt import create_react_agent

This project always uses the LangGraph version.

Error 4: Path traversal in read_file not detected

Symptom: The agent can read files outside the sandbox.

Solution: Always use os.path.normpath + startswith:

safe_path = os.path.normpath(os.path.join(SANDBOX_DIR, file_path))
if not safe_path.startswith(SANDBOX_DIR):
    return "Security error: access denied."

Without normpath, a path like ../../../etc/passwd could slip past the validation.

Error 5: Pydantic ValidationError crashes the agent

Symptom: A Pydantic exception when the model passes an invalid input.

Cause: LangGraph's create_react_agent catches ValidationError automatically and turns it into a ToolMessage with the error. If you use the manual execution loop, you need to catch it yourself:

from pydantic import ValidationError

try:
    result = my_tool.invoke(tool_call["args"])
except ValidationError as e:
    result = f"Validation error: {e}"

With create_react_agent, this is handled automatically.

Error 6: The agent doesn't use read_file because it doesn't know which files exist

Symptom: The user asks "what does the notes file say?" and the agent doesn't even try the tool.

Solution: Improve the description to include example files, or tell it that if it doesn't know the name, it should try a likely one — the error will tell it which files are available.

---

Project Resources

1. Open-Meteo API Documentation — Free weather API. Includes geocoding, forecast, historical data. No key.
2. DuckDuckGo Search Python Package — The duckduckgo-search package. Methods for text search, news, images.
3. LangGraph create_react_agent Reference — API reference for the prebuilt agent.
4. Pydantic Field Validators — Validator documentation: field_validator, model_validator, constraints.
5. Python AST Module — Reference for the ast module for safe expression parsing.
6. LangChain Custom Tools Guide — Official guide to creating tools with @tool and args_schema.

---

Connection to the Next Module

In this project you built 5 tools with real APIs, Pydantic schemas, and error handling. The agent combines them with create_react_agent to resolve multi-tool queries. It's a solid agent — but it operates in a single pattern: the model decides which tools to use, executes them, and answers.

In Module 3 (Function Calling Patterns) you'll take tool calling to another level:

• Parallel function calling: Force the model to call multiple tools at once
• Tool routing: Direct queries to specific subsets of tools based on the request type
• Structured extraction: Use function calling to extract structured data from text
• Composition patterns: Chain one tool's result as the input to another
• Retry and circuit breaker: Retry automatically with backoff, or disable tools temporarily

The 5 tools you built here are your base toolkit. In Module 3, the focus shifts from "how do I build solid tools?" to "how do I orchestrate tools like a professional?" And when you start the evolving project in Module 4, the web search and file reading tools become the Research Agent's tools — the central project that evolves into a production-ready multi-agent system.

---