Module 3: Tool Calling & Function Use

Giving Agents Real-World Capabilities

---

Introduction: From Thinking to Doing

In Modules 1 and 2, you learned what agents are and how to build chains with LangChain. But chains are still linear — input goes in, output comes out. A real agent needs to interact with the outside world.

Tool calling is the mechanism that transforms a language model from a text generator into an agent that can fetch data, call APIs, query databases, and manipulate files. This module is where your agents start doing real work.

By the end of this module, you will:

• Define tools using Pydantic schemas for type safety
• Use the @tool decorator for rapid tool creation
• Connect agents to external APIs and data sources
• Handle tool errors gracefully with retry strategies
• Build a working web search agent using Tavily

---

3.1 Defining Tools with Pydantic Schemas

Why Type Safety Matters for Tools

When an LLM calls a tool, it generates the function arguments as JSON. Without validation, anything could go wrong:

• Missing required parameters
• Wrong data types (string instead of integer)
• Values outside expected ranges

Pydantic solves this by validating every tool input before your function executes.

Pydantic Basics for Tool Schemas

from pydantic import BaseModel, Field

class WeatherQuery(BaseModel):
    """Schema for weather lookup tool."""
    city: str = Field(description="The city name to look up weather for")
    units: str = Field(
        default="celsius",
        description="Temperature units: 'celsius' or 'fahrenheit'"
    )

# Pydantic validates automatically
query = WeatherQuery(city="Tokyo", units="celsius")
print(query.city)   # "Tokyo"
print(query.units)  # "celsius"

# Invalid input raises a clear error
try:
    bad_query = WeatherQuery(units="celsius")  # Missing required 'city'
except Exception as e:
    print(f"Validation error: {e}")

Defining a LangChain Tool with Pydantic

LangChain tools accept a Pydantic model as their args_schema, giving you automatic validation and documentation:

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional


class StockPriceInput(BaseModel):
    """Input schema for stock price lookup."""
    symbol: str = Field(description="The stock ticker symbol (e.g., AAPL, GOOGL)")
    include_history: bool = Field(
        default=False,
        description="Whether to include 30-day price history"
    )


class StockPriceTool(BaseTool):
    name: str = "get_stock_price"
    description: str = "Look up the current stock price for a given ticker symbol"
    args_schema: type = StockPriceInput

    def _run(self, symbol: str, include_history: bool = False) -> str:
        """Execute the stock price lookup."""
        # In production, this would call a real API
        prices = {
            "AAPL": 178.50,
            "GOOGL": 142.30,
            "MSFT": 415.20,
            "TSLA": 248.90,
        }

        price = prices.get(symbol.upper())
        if price is None:
            return f"Stock symbol '{symbol}' not found."

        result = f"{symbol.upper()}: ${price:.2f}"

        if include_history:
            result += " (30-day change: +3.2%)"

        return result


# Create an instance of the tool
stock_tool = StockPriceTool()

# Test it directly
print(stock_tool.invoke({"symbol": "AAPL"}))
# "AAPL: $178.50"

print(stock_tool.invoke({"symbol": "AAPL", "include_history": True}))
# "AAPL: $178.50 (30-day change: +3.2%)"

---

3.2 The @tool Decorator

Rapid Tool Creation

The class-based approach is thorough, but LangChain provides a much faster way to create tools — the @tool decorator. It automatically generates the schema from your function's type hints and docstring.

from langchain_core.tools import tool


@tool
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city.

    Args:
        city: The city name to look up weather for.
        units: Temperature units, either 'celsius' or 'fahrenheit'.
    """
    # Simulated weather data
    weather_data = {
        "Tokyo": {"temp_c": 18, "condition": "Rainy"},
        "London": {"temp_c": 12, "condition": "Cloudy"},
        "New York": {"temp_c": 25, "condition": "Sunny"},
        "Paris": {"temp_c": 20, "condition": "Partly Cloudy"},
    }

    data = weather_data.get(city)
    if data is None:
        return f"Weather data not available for '{city}'."

    temp = data["temp_c"]
    if units == "fahrenheit":
        temp = temp * 9 / 5 + 32
        unit_label = "F"
    else:
        unit_label = "C"

    return f"{city}: {temp}{unit_label}, {data['condition']}"


# The decorator creates a fully functional LangChain tool
print(get_weather.name)         # "get_weather"
print(get_weather.description)  # "Get the current weather for a city."

# Invoke it
print(get_weather.invoke({"city": "Tokyo"}))
# "Tokyo: 18C, Rainy"

Combining @tool with Pydantic for Richer Schemas

When you need more control over validation, combine the @tool decorator with a Pydantic args_schema:

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


class CalculatorInput(BaseModel):
    """Input for the calculator tool."""
    expression: str = Field(
        description="A mathematical expression to evaluate (e.g., '2 + 2', '100 * 0.15')"
    )
    precision: int = Field(
        default=2,
        description="Number of decimal places in the result",
        ge=0,
        le=10
    )


@tool(args_schema=CalculatorInput)
def calculator(expression: str, precision: int = 2) -> str:
    """Evaluate a mathematical expression and return the result."""
    try:
        # Safety check: only allow math characters
        allowed = set("0123456789+-*/.() ")
        if not all(c in allowed for c in expression):
            return "Error: Expression contains invalid characters."

        result = eval(expression)
        return f"{expression} = {result:.{precision}f}"
    except ZeroDivisionError:
        return "Error: Division by zero."
    except Exception as e:
        return f"Error: {str(e)}"


print(calculator.invoke({"expression": "100 * 0.15", "precision": 4}))
# "100 * 0.15 = 15.0000"

Creating Multiple Tools for an Agent

In practice, agents have access to multiple tools. Here is how you define a toolkit:

from langchain_core.tools import tool
from datetime import datetime


@tool
def get_current_time(timezone: str = "UTC") -> str:
    """Get the current date and time.

    Args:
        timezone: The timezone (currently only supports 'UTC').
    """
    now = datetime.utcnow()
    return f"Current time (UTC): {now.strftime('%Y-%m-%d %H:%M:%S')}"


@tool
def search_contacts(name: str) -> str:
    """Search for a contact by name and return their information.

    Args:
        name: The name of the contact to search for.
    """
    contacts = {
        "Alice": {"email": "alice@example.com", "phone": "555-0101"},
        "Bob": {"email": "bob@example.com", "phone": "555-0102"},
        "Charlie": {"email": "charlie@example.com", "phone": "555-0103"},
    }

    contact = contacts.get(name)
    if contact:
        return f"{name}: email={contact['email']}, phone={contact['phone']}"
    return f"No contact found with name '{name}'."


@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a recipient.

    Args:
        to: The recipient's email address.
        subject: The email subject line.
        body: The email body text.
    """
    # In production, this would use an email API
    return f"Email sent to {to} with subject '{subject}'."


# Collect all tools into a list
tools = [get_weather, get_current_time, search_contacts, send_email, calculator]

---

3.3 Connecting Agents to APIs, Databases, and Files

API Integration

Real-world tools connect to external APIs. Here is a pattern for building API-backed tools:

import os
import httpx
from langchain_core.tools import tool


@tool
def lookup_company(company_name: str) -> str:
    """Look up basic information about a company.

    Args:
        company_name: The name of the company to look up.
    """
    # Example using a hypothetical API
    # In production, you would call a real API like Clearbit, Crunchbase, etc.
    api_key = os.getenv("COMPANY_API_KEY")

    try:
        response = httpx.get(
            "https://api.example.com/companies/search",
            params={"name": company_name},
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10.0
        )
        response.raise_for_status()
        data = response.json()

        return (
            f"Company: {data['name']}\n"
            f"Industry: {data['industry']}\n"
            f"Employees: {data['employee_count']}\n"
            f"Founded: {data['founded_year']}"
        )
    except httpx.TimeoutException:
        return f"Error: Request timed out while looking up '{company_name}'."
    except httpx.HTTPStatusError as e:
        return f"Error: API returned status {e.response.status_code}."
    except Exception as e:
        return f"Error looking up company: {str(e)}"

Database Integration

Agents can query databases to retrieve or update information:

import sqlite3
from langchain_core.tools import tool


@tool
def query_orders(customer_email: str) -> str:
    """Look up orders for a customer by their email address.

    Args:
        customer_email: The customer's email address.
    """
    try:
        conn = sqlite3.connect("orders.db")
        cursor = conn.cursor()

        cursor.execute(
            """
            SELECT order_id, product, status, order_date
            FROM orders
            WHERE customer_email = ?
            ORDER BY order_date DESC
            LIMIT 5
            """,
            (customer_email,)
        )

        rows = cursor.fetchall()
        conn.close()

        if not rows:
            return f"No orders found for '{customer_email}'."

        results = []
        for row in rows:
            results.append(
                f"Order #{row[0]}: {row[1]} — Status: {row[2]} ({row[3]})"
            )

        return f"Orders for {customer_email}:\n" + "\n".join(results)

    except Exception as e:
        return f"Error querying orders: {str(e)}"

File System Integration

Agents can read and write files when they need to process documents:

import os
from langchain_core.tools import tool


@tool
def read_file(file_path: str) -> str:
    """Read the contents of a text file.

    Args:
        file_path: The path to the file to read.
    """
    # Security: restrict to a safe directory
    safe_dir = os.path.abspath("./workspace")
    full_path = os.path.abspath(os.path.join(safe_dir, file_path))

    if not full_path.startswith(safe_dir):
        return "Error: Access denied. File is outside the workspace directory."

    try:
        with open(full_path, "r") as f:
            content = f.read()
        return f"Contents of {file_path}:\n{content}"
    except FileNotFoundError:
        return f"Error: File '{file_path}' not found."
    except Exception as e:
        return f"Error reading file: {str(e)}"


@tool
def write_file(file_path: str, content: str) -> str:
    """Write content to a text file in the workspace directory.

    Args:
        file_path: The path to the file to write.
        content: The content to write to the file.
    """
    safe_dir = os.path.abspath("./workspace")
    full_path = os.path.abspath(os.path.join(safe_dir, file_path))

    if not full_path.startswith(safe_dir):
        return "Error: Access denied. File is outside the workspace directory."

    try:
        os.makedirs(os.path.dirname(full_path), exist_ok=True)
        with open(full_path, "w") as f:
            f.write(content)
        return f"Successfully wrote {len(content)} characters to {file_path}."
    except Exception as e:
        return f"Error writing file: {str(e)}"

---

3.4 Tool Error Handling and Retry Strategies

Why Error Handling Matters

Tool calls fail. APIs time out, databases go down, rate limits get hit. A robust agent handles these failures gracefully instead of crashing.

Pattern 1: Return Errors as Strings

The simplest strategy — return error messages that the LLM can reason about:

@tool
def fetch_data(url: str) -> str:
    """Fetch data from a URL.

    Args:
        url: The URL to fetch data from.
    """
    try:
        response = httpx.get(url, timeout=10.0)
        response.raise_for_status()
        return response.text[:2000]  # Truncate to avoid token limits
    except httpx.TimeoutException:
        return "Error: The request timed out. The server may be slow or unavailable."
    except httpx.HTTPStatusError as e:
        return f"Error: HTTP {e.response.status_code}. The URL may be invalid or the server is returning an error."
    except Exception as e:
        return f"Error: {str(e)}"

When the agent receives an error string, it can reason about what to do next — try a different URL, use a different tool, or inform the user.

Pattern 2: Retry with Backoff

For transient failures, implement automatic retries:

import time
from langchain_core.tools import tool


def retry_with_backoff(func, max_retries: int = 3, base_delay: float = 1.0):
    """Decorator that retries a function with exponential backoff."""
    def wrapper(*args, **kwargs):
        for attempt in range(max_retries):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if attempt == max_retries - 1:
                    return f"Error after {max_retries} attempts: {str(e)}"
                delay = base_delay * (2 ** attempt)
                print(f"  Retry {attempt + 1}/{max_retries} in {delay}s: {e}")
                time.sleep(delay)
        return "Error: Max retries exceeded."
    return wrapper


@tool
def reliable_api_call(endpoint: str) -> str:
    """Call an API endpoint with automatic retry on failure.

    Args:
        endpoint: The API endpoint path to call.
    """
    @retry_with_backoff
    def _call_api(endpoint: str) -> str:
        response = httpx.get(
            f"https://api.example.com/{endpoint}",
            timeout=10.0
        )
        response.raise_for_status()
        return response.text

    return _call_api(endpoint)

Pattern 3: Fallback Tools

Provide alternative tools that the agent can use when the primary tool fails:

@tool
def search_web_primary(query: str) -> str:
    """Search the web using the primary search API (Tavily).

    Args:
        query: The search query.
    """
    try:
        # Primary search API
        results = tavily_client.search(query)
        return format_results(results)
    except Exception as e:
        return f"Primary search failed: {str(e)}. Try using search_web_backup."


@tool
def search_web_backup(query: str) -> str:
    """Backup web search tool. Use this if the primary search fails.

    Args:
        query: The search query.
    """
    try:
        # Backup search API
        response = httpx.get(
            "https://api.duckduckgo.com/",
            params={"q": query, "format": "json"}
        )
        return response.text[:2000]
    except Exception as e:
        return f"Backup search also failed: {str(e)}"

The agent's system prompt can instruct it to try the backup when the primary fails. This is one of the key advantages of agentic systems — they can adapt to failures.

---

3.5 Interactive Exercise: Build a Web Search Agent

The Goal

Build an agent that can search the web using Tavily, analyze the results, and provide a synthesized answer. This is a practical agent pattern used in production research tools.

Setting Up Tavily

Tavily is an AI-optimized search API designed specifically for agent use. Sign up at tavily.com for a free API key.

Add to your .env file:

TAVILY_API_KEY=tvly-your-key-here

The Complete Agent

Create search_agent.py:

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.prebuilt import create_react_agent

load_dotenv()


# Define the search tool
@tool
def web_search(query: str) -> str:
    """Search the web for current information on any topic.

    Args:
        query: The search query to look up.
    """
    try:
        from tavily import TavilyClient

        client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
        response = client.search(query, max_results=5)

        results = []
        for item in response.get("results", []):
            results.append(
                f"Title: {item['title']}\n"
                f"URL: {item['url']}\n"
                f"Content: {item['content'][:300]}\n"
            )

        if not results:
            return "No results found for the query."

        return "\n---\n".join(results)

    except Exception as e:
        return f"Search error: {str(e)}"


@tool
def summarize_text(text: str, max_sentences: int = 3) -> str:
    """Summarize a long text into a shorter version.

    Args:
        text: The text to summarize.
        max_sentences: Maximum number of sentences in the summary.
    """
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    response = llm.invoke(
        f"Summarize the following text in {max_sentences} sentences:\n\n{text}"
    )
    return response.content


# Create the agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [web_search, summarize_text]

agent = create_react_agent(
    model=llm,
    tools=tools,
)

# System message for the agent
system_message = SystemMessage(
    content=(
        "You are a research assistant. When asked a question:\n"
        "1. Search the web for relevant, current information\n"
        "2. Analyze and synthesize the results\n"
        "3. Provide a clear, well-structured answer with sources\n"
        "4. If the search results are insufficient, search again with different terms\n"
        "Always cite your sources."
    )
)


def research(question: str) -> str:
    """Run a research query through the agent."""
    print(f"\nResearching: {question}\n")

    result = agent.invoke({
        "messages": [system_message, HumanMessage(content=question)]
    })

    # Get the final response
    final_message = result["messages"][-1]
    return final_message.content


def main():
    print("Web Search Agent")
    print("Type your research question (or 'exit' to quit).\n")

    while True:
        question = input("Question: ")

        if question.lower() == "exit":
            print("Goodbye!")
            break

        answer = research(question)
        print(f"\nAnswer:\n{answer}\n")
        print("-" * 60)


if __name__ == "__main__":
    main()

Run It

# python search_agent.py

Try These Research Queries

Question: What are the latest developments in AI agent frameworks?
Question: Compare LangChain and CrewAI for building multi-agent systems.
Question: What is the current state of autonomous AI agents in customer service?

What You Built

This search agent demonstrates several key patterns:

1. Tool Definition: The web_search tool connects to an external API with proper error handling
2. ReACT Agent: create_react_agent implements the full ReACT loop automatically
3. Multi-Tool Usage: The agent can search the web and then summarize results
4. Iterative Research: If initial results are insufficient, the agent can search again with different terms
5. Source Citation: The system prompt instructs the agent to cite its sources

This is the same pattern used by tools like Perplexity, ChatGPT with browsing, and enterprise research assistants.

---

Key Takeaways

1. Pydantic schemas provide type-safe, validated tool inputs that prevent runtime errors
2. The @tool decorator is the fastest way to create LangChain tools from regular Python functions
3. Tool integrations with APIs, databases, and files give agents real-world capabilities
4. Error handling is essential — return errors as strings so the agent can reason about them and adapt
5. Tavily + create_react_agent gives you a production-ready research agent in under 50 lines of code

---

Exercises

Before moving to Module 4, try these challenges:

1. Add a calculator tool to the search agent so it can perform calculations on data it finds (e.g., "What is the market cap of Apple and Microsoft combined?").

2. Build a multi-source research agent that has tools for web search, Wikipedia lookup, and news search. Let the agent decide which source to use based on the question.

3. Implement rate limiting: Add a rate limiter to the web_search tool that enforces a maximum of 5 searches per minute to avoid hitting API limits.

4. Add a "save results" tool that writes the agent's research to a markdown file. Ask the agent to research a topic and save a report.

5. Error recovery experiment: Deliberately break one of your tools (e.g., use an invalid API key) and observe how the agent handles the error. Then add a fallback tool and test again.

---

Next up: Module 4, where we build stateful agents with LangGraph — conditional routing, human-in-the-loop patterns, and production-ready workflows.