Tools Are the Whole Point

An agent without tools is a chatbot with delusions of grandeur. Tools are what separate "I can tell you about sending an email" from "I just sent the email." Every real-world agent capability — database queries, API calls, file operations, web browsing — is implemented as a tool.

This lesson goes beyond basic tool definition (covered in Lesson 3) into the patterns that make tools reliable, secure, and composable in production.

Custom Tool Patterns

Pattern 1: The Structured Tool

For complex tools with multiple parameters and validation, use Pydantic schemas:

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

class DatabaseQueryInput(BaseModel):
    """Input for querying the product database."""
    query: str = Field(description="SQL-safe search term")
    category: Optional[str] = Field(
        default=None,
        description="Product category filter",
        json_schema_extra={"enum": ["electronics", "books", "clothing"]}
    )
    limit: int = Field(default=10, ge=1, le=100, description="Max results")

def search_products(query: str, category: str = None, limit: int = 10) -> str:
    """Search the product database and return matching items."""
    # Real implementation: parameterized SQL query
    import sqlite3
    conn = sqlite3.connect("products.db")
    sql = "SELECT name, price, stock FROM products WHERE name LIKE ?"
    params = [f"%{query}%"]
    if category:
        sql += " AND category = ?"
        params.append(category)
    sql += " LIMIT ?"
    params.append(limit)
    results = conn.execute(sql, params).fetchall()
    return str(results)

product_search = StructuredTool.from_function(
    func=search_products,
    name="search_products",
    description="Search the product database by name. Optionally filter by category.",
    args_schema=DatabaseQueryInput
)

The Pydantic schema gives you free input validation. If the LLM passes limit=500, the tool rejects it before execution. That's a guardrail that costs nothing.

Pattern 2: The Async Tool

For tools that call external APIs or do I/O, async is non-negotiable in production:

from langchain_core.tools import tool
import httpx

@tool
async def fetch_github_repo(owner: str, repo: str) -> str:
    """Get information about a GitHub repository including stars, forks, and description."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.github.com/repos/{owner}/{repo}",
            headers={"Accept": "application/vnd.github.v3+json"}
        )
        if resp.status_code == 404:
            return f"Repository {owner}/{repo} not found"
        if resp.status_code != 200:
            return f"GitHub API error: {resp.status_code}"
        data = resp.json()
        return (
            f"Name: {data['full_name']}\n"
            f"Stars: {data['stargazers_count']:,}\n"
            f"Forks: {data['forks_count']:,}\n"
            f"Description: {data['description']}\n"
            f"Language: {data['language']}\n"
            f"Last updated: {data['updated_at']}"
        )

Pattern 3: The Tool with Side Effects Tracking

Some tools modify state — sending emails, creating database records, deploying code. These need explicit tracking:

from datetime import datetime

# Side effect log — in production, this is a database table
side_effects: list[dict] = []

@tool
def send_notification(recipient: str, subject: str, body: str) -> str:
    """Send an email notification. This action cannot be undone.
    Only use when explicitly asked to send a message."""
    # Log the side effect BEFORE executing
    effect = {
        "type": "email",
        "recipient": recipient,
        "subject": subject,
        "timestamp": datetime.now().isoformat(),
        "status": "pending"
    }
    side_effects.append(effect)

    try:
        # Actually send the email
        result = email_client.send(to=recipient, subject=subject, body=body)
        effect["status"] = "sent"
        effect["message_id"] = result.id
        return f"Email sent to {recipient}. Message ID: {result.id}"
    except Exception as e:
        effect["status"] = "failed"
        effect["error"] = str(e)
        return f"Failed to send email: {str(e)}"

Notice the tool description includes "This action cannot be undone." This signals the LLM to be cautious. Prompt engineering in tool descriptions is just as important as in system prompts.

MCP: The Universal Tool Protocol

Anthropic's Model Context Protocol standardizes how agents connect to tools. Instead of writing custom tool wrappers for every service, MCP servers expose tools through a universal protocol that any MCP-compatible client can discover and use.

# Using MCP tools with LangChain
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

async def build_mcp_agent():
    """Build an agent that uses MCP servers as tools."""
    async with MultiServerMCPClient(
        {
            "filesystem": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/workspace"],
                "transport": "stdio"
            },
            "github": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-github"],
                "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxx"},
                "transport": "stdio"
            }
        }
    ) as client:
        # All MCP tools are automatically available
        tools = client.get_tools()
        llm = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(llm, tools)

        result = await agent.ainvoke({
            "messages": [{"role": "user", "content": "List files in /tmp/workspace and check if any are tracked in a GitHub repo"}]
        })
        return result

MCP is the future of agent tool integration. Instead of maintaining a custom tool for every service, you use MCP servers — pre-built, standardized, and maintained by the community. Browse available servers at repos.skila.ai.

Tool Security: The Permission Model

Tools are the most dangerous part of any agent system. A tool with database write access, email sending capability, or code execution is a security risk. Here's the defense-in-depth model:

Layer 1: Least-Privilege Tools

# BAD: One tool does everything
@tool
def database(action: str, sql: str) -> str:
    """Execute any SQL query."""
    return db.execute(sql)  # Can DROP TABLE

# GOOD: Separate read and write tools
@tool
def search_database(query: str, table: str) -> str:
    """Read-only search. Cannot modify data."""
    allowed_tables = ["products", "categories"]
    if table not in allowed_tables:
        return f"Access denied: table {table} is not searchable"
    return db.execute(f"SELECT * FROM {table} WHERE name LIKE ?", [f"%{query}%"])

@tool
def update_product_stock(product_id: str, new_stock: int) -> str:
    """Update stock count for a specific product. Requires confirmation."""
    if new_stock < 0:
        return "Error: stock cannot be negative"
    # ... update with audit log

Layer 2: Input Validation

from pydantic import BaseModel, Field, field_validator

class SafeSearchInput(BaseModel):
    query: str = Field(max_length=200)
    table: str = Field(pattern=r"^[a-z_]+$")  # Only lowercase + underscore

    @field_validator("query")
    @classmethod
    def no_sql_injection(cls, v: str) -> str:
        dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "--", ";"]
        if any(d in v.upper() for d in dangerous):
            raise ValueError("Query contains potentially dangerous SQL keywords")
        return v

Layer 3: Human-in-the-Loop for Dangerous Actions

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

# LangGraph's interrupt_before pauses execution for human approval
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("dangerous_tools", dangerous_tool_node)
graph.add_node("safe_tools", safe_tool_node)

# Interrupt before executing dangerous tools
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["dangerous_tools"]  # Pauses here for human approval
)

This creates a checkpoint right before dangerous actions execute. The agent pauses, shows you what it wants to do, and waits for your approval before proceeding.

Tool Chaining: Composable Actions

Advanced agents combine multiple tools in sequence. The key is making tool outputs compatible with other tools' inputs:

@tool
def search_issues(repo: str, query: str) -> str:
    """Search GitHub issues. Returns issue numbers and titles."""
    # Returns: "#123: Bug in login flow\n#456: Performance regression"
    ...

@tool
def get_issue_details(repo: str, issue_number: int) -> str:
    """Get full details of a GitHub issue including comments."""
    ...

@tool
def create_branch(repo: str, branch_name: str, base: str = "main") -> str:
    """Create a new git branch for working on a fix."""
    ...

# The agent chains these naturally:
# 1. search_issues("myorg/myrepo", "login timeout")
# 2. get_issue_details("myorg/myrepo", 123)  # From search results
# 3. create_branch("myorg/myrepo", "fix/login-timeout-123")

You don't need to explicitly define the chain. The agent reads each tool's output and decides which tool to call next. That's the power of the ReAct pattern — the LLM handles composition dynamically.

Error Handling: Tools Will Fail

APIs timeout. Databases go down. Rate limits hit. Your tools will fail, and the agent needs to handle it gracefully:

@tool
def robust_web_search(query: str) -> str:
    """Search the web. Returns results or a clear error message if search fails."""
    try:
        results = tavily.search(query=query, max_results=5)
        if not results.get("results"):
            return f"No results found for '{query}'. Try a different search term."
        return format_results(results)
    except httpx.TimeoutException:
        return "Search timed out. The search service may be slow. Try again with a simpler query."
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 429:
            return "Search rate limit reached. Wait a moment before searching again."
        return f"Search service error (HTTP {e.response.status_code}). Try again later."
    except Exception as e:
        return f"Search failed: {str(e)}. You can try an alternative approach."

The error messages are designed for the LLM to read and act on. "Try a different search term" gives the agent a recovery strategy. "Wait a moment" tells it to try other tools first. The agent can reason about errors and adapt — but only if the error messages are actionable.

Tool Use — Giving Agents Real-World Capabilities

Related Resources