Tool Use in Agents

Creating Custom Tools

14m read

Creating Custom Tools

Custom tools are what give your agent domain-specific capabilities. A well-designed tool is reliable, has a clear and unambiguous interface, handles errors gracefully, and comes with a description that the LLM can understand and use correctly. This lesson walks through building production-quality custom tools.

The Anatomy of a Good Tool

A great tool has four qualities:

  1. Single responsibility: Does one thing well. search_database is better than do_stuff_with_data.
  2. Clear description: The docstring tells the LLM when to use it and what it does — not just its signature.
  3. Reliable output: Returns a consistent, predictable structure. Never returns None silently.
  4. Graceful error handling: Returns error information as a string, not an exception (which would crash the agent loop).

Building Tools with LangChain's @tool Decorator

from langchain_core.tools import tool
from typing import Optional
import httpx
import json

@tool
def fetch_github_repo_info(owner: str, repo: str) -> str:
    """Fetch metadata about a GitHub repository.
    
    Use this when you need information about a GitHub repository such as 
    its description, star count, last commit date, main language, or open issues.
    
    Args:
        owner: The GitHub username or organization name (e.g., 'langchain-ai')
        repo: The repository name (e.g., 'langchain')
    
    Returns:
        A JSON string with repository metadata, or an error message.
    """
    try:
        response = httpx.get(
            f"https://api.github.com/repos/{owner}/{repo}",
            headers={"Accept": "application/vnd.github.v3+json"},
            timeout=10.0,
        )
        response.raise_for_status()
        data = response.json()
        
        # Return only relevant fields (don't dump the full 100-field API response)
        return json.dumps({
            "name": data["name"],
            "description": data.get("description", "No description"),
            "stars": data["stargazers_count"],
            "forks": data["forks_count"],
            "language": data.get("language", "Unknown"),
            "open_issues": data["open_issues_count"],
            "last_pushed": data["pushed_at"],
            "license": data.get("license", {}).get("name", "No license"),
            "url": data["html_url"],
        }, indent=2)
    
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return f"Repository {owner}/{repo} not found. Check the owner and repo name."
        return f"GitHub API error: {e.response.status_code} — {e.response.text[:200]}"
    except httpx.TimeoutException:
        return "GitHub API request timed out after 10 seconds. Try again."
    except Exception as e:
        return f"Unexpected error fetching repo info: {str(e)}"

Building Tools with Pydantic Input Validation

For tools with complex inputs, use Pydantic for validation:

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field, field_validator
from typing import Type, Any

class DatabaseQueryInput(BaseModel):
    table: str = Field(description="Table name to query (snake_case)")
    filters: dict[str, Any] = Field(default_factory=dict, description="Column-value filter pairs")
    columns: list[str] = Field(default_factory=list, description="Columns to return. Empty list returns all.")
    limit: int = Field(default=10, ge=1, le=100, description="Max rows to return (1-100)")
    
    @field_validator("table")
    @classmethod
    def validate_table_name(cls, v: str) -> str:
        # Prevent SQL injection in table name
        allowed = set("abcdefghijklmnopqrstuvwxyz_0123456789")
        if not all(c in allowed for c in v.lower()):
            raise ValueError(f"Invalid table name: {v}. Use only letters, numbers, and underscores.")
        return v.lower()


class DatabaseQueryTool(BaseTool):
    name: str = "query_database"
    description: str = """Query the product database for information about products, orders, or customers.
    
    Use this when you need to look up specific data from the database.
    Supports filtering by any column and limiting the number of results.
    Never use this for write operations — it is read-only.
    """
    args_schema: Type[BaseModel] = DatabaseQueryInput
    
    def _run(self, table: str, filters: dict = None, columns: list = None, limit: int = 10) -> str:
        """Execute a safe read-only database query."""
        filters = filters or {}
        columns = columns or []
        
        try:
            # Build safe parameterized query
            selected_cols = ", ".join(columns) if columns else "*"
            query = f"SELECT {selected_cols} FROM {table}"
            params = []
            
            if filters:
                conditions = [f"{col} = ?" for col in filters.keys()]
                query += " WHERE " + " AND ".join(conditions)
                params = list(filters.values())
            
            query += f" LIMIT {limit}"
            
            # Execute (using your actual database connection)
            # results = db.execute(query, params).fetchall()
            # For demo:
            results = [{"id": 1, "name": "Example", "status": "active"}]
            
            return json.dumps({"rows": results, "count": len(results), "query_info": {"table": table, "filters": filters}})
        
        except Exception as e:
            return f"Database query failed: {str(e)}"
    
    async def _arun(self, **kwargs) -> str:
        """Async version for use in async agent contexts."""
        # Wrap sync in executor for true async
        import asyncio
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, lambda: self._run(**kwargs))

Tool Description Best Practices

The description is everything — the LLM reads it to decide when and how to use the tool:

# BAD: Vague, doesn't explain when to use it
@tool
def search(query: str) -> str:
    """Search for something."""

# GOOD: Specific, includes use cases and limitations
@tool  
def search_product_catalog(query: str, category: Optional[str] = None, max_price: Optional[float] = None) -> str:
    """Search the product catalog for items matching the query.
    
    Use this when the user asks about available products, wants to find items by name,
    description, or category, or asks 'do you have X?'.
    
    DO NOT use this for checking order status, customer accounts, or pricing history.
    
    Args:
        query: Natural language search query (e.g., 'wireless headphones' or 'red shoes size 10')
        category: Optional category filter ('electronics', 'clothing', 'books', etc.)
        max_price: Optional maximum price in USD
    
    Returns:
        JSON array of matching products with id, name, price, and availability.
        Returns empty array if no products match.
    """

Tool Registry Pattern

For agents with many tools, a registry pattern keeps things organized:

from langchain_core.tools import BaseTool

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, BaseTool] = {}
    
    def register(self, tool: BaseTool) -> None:
        self._tools[tool.name] = tool
    
    def get_tools_for_context(self, context: str) -> list[BaseTool]:
        """Return only tools relevant to the current context."""
        # Filter tools based on context (e.g., only database tools for DB questions)
        return list(self._tools.values())
    
    def all_tools(self) -> list[BaseTool]:
        return list(self._tools.values())

registry = ToolRegistry()
registry.register(DatabaseQueryTool())
# registry.register(OtherTool())

The investment in good tool design pays dividends: a well-described, reliable tool is used correctly by the model nearly every time, while a poorly designed one causes constant agent failures.