Build Your First AI Agent from Scratch - Part 4: Implementing Tool Usage and API Integrations

Welcome to Part 4 of our comprehensive AI agent tutorial series! Your agent now has memory, context handling, and conversation capabilities. It's time to transform it from a conversational assistant into a powerful action-taking agent by adding tool usage and external API integrations.

This is where your agent becomes truly useful. Instead of just talking, it can now search the web, retrieve data, interact with external services, and perform real-world tasks on behalf of users.

What You'll Learn in This Tutorial

By the end of this tutorial, you'll have:

• ✅ Flexible tool system that can be easily extended with new capabilities
• ✅ Web search integration for real-time information retrieval
• ✅ API integration framework for connecting to external services
• ✅ Error handling and retry logic for robust tool usage
• ✅ Tool selection intelligence so your agent knows when to use which tool
• ✅ Comprehensive testing for all tool functionality

Estimated Time: 30-35 minutes

---

Step 1: Understanding Tool Architecture for AI Agents

Before we write code, it helps to map the terrain. Tools are the agent’s hands and eyes — how it reaches out to act.

Before diving into implementation, let's understand why tools are crucial for AI agents and how to design them effectively.

Why AI Agents Need Tools

Limited Knowledge Cutoff: Language models have training data cutoffs and can't access real-time information.

No Action Capability: Pure language models can only generate text, not perform actions in the real world.

Hallucination Issues: Without access to real data, models may generate plausible but incorrect information.

Dynamic Information Needs: Users often need current data (weather, stock prices, news) that requires live API calls.

Tool Design Principles

Modularity: Each tool should have a single, well-defined responsibility.

Standardized Interface: All tools should implement the same interface for consistency.

Error Resilience: Tools must handle failures gracefully and provide meaningful error messages.

Composability: Tools should work together and be easily combined.

Testability: Each tool should be independently testable.

Tool Architecture Overview

Our tool system will use:

1. Base Tool Class - Defines the standard interface
2. Concrete Tool Implementations - Specific functionality (web search, APIs, etc.)
3. Tool Manager - Handles tool selection and execution
4. Integration Layer - Connects tools to the agent's decision-making process

---

Step 2: Building the Tool Foundation

Let’s establish a tiny, consistent interface so every tool feels identical to the agent.

Let's start by creating a robust foundation for our tool system.

Base Tool Interface

Create src/tools/base_tool.py:

"""
Base tool interface for AI Agent tools

This module defines the standard interface that all tools must implement,
providing consistency and enabling the agent to use any tool through the same API.
"""

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class ToolResult:
    """
    Standardized result from tool execution
    
    This class ensures all tools return results in a consistent format,
    making it easier for the agent to process and understand tool outputs.
    """
    success: bool                           # Whether the tool executed successfully
    content: str                           # The main result content
    metadata_json: Dict[str, Any]               # Additional information about the execution
    error_message: Optional[str] = None    # Error details if success=False
    execution_time: Optional[float] = None # How long the tool took to execute
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert result to dictionary for serialization"""
        return {
            "success": self.success,
            "content": self.content,
            "metadata_json": self.metadata_json,
            "error_message": self.error_message,
            "execution_time": self.execution_time
        }
    
    def to_json(self) -> str:
        """Convert result to JSON string"""
        return json.dumps(self.to_dict(), default=str)

class BaseTool(ABC):
    """
    Abstract base class for all AI agent tools
    
    This class defines the interface that all tools must implement.
    It provides common functionality and ensures consistency across tools.
    """
    
    def __init__(self, name: str, description: str):
        """
        Initialize the tool
        
        Args:
            name: Unique name for the tool
            description: Human-readable description of what the tool does
        """
        self.name = name
        self.description = description
        self.usage_count = 0
        self.last_used = None
        self.is_enabled = True
    
    @abstractmethod
    def execute(self, query: str, **kwargs) -> ToolResult:
        """
        Execute the tool with the given query
        
        This is the main method that subclasses must implement.
        It should perform the tool's specific functionality and return a ToolResult.
        
        Args:
            query: The input query or command for the tool
            **kwargs: Additional parameters specific to the tool
            
        Returns:
            ToolResult: Standardized result object
        """
        pass
    
    @abstractmethod
    def get_parameters(self) -> Dict[str, Any]:
        """
        Get the parameters this tool accepts
        
        Returns a dictionary describing the parameters this tool can accept,
        including types, descriptions, and whether they're required.
        
        Returns:
            Dict describing the tool's parameters
        """
        pass
    
    def validate_parameters(self, **kwargs) -> bool:
        """
        Validate parameters before execution
        
        Args:
            **kwargs: Parameters to validate
            
        Returns:
            True if parameters are valid, False otherwise
        """
        # Default implementation - can be overridden by subclasses
        return True
    
    def get_usage_stats(self) -> Dict[str, Any]:
        """Get usage statistics for this tool"""
        return {
            "name": self.name,
            "usage_count": self.usage_count,
            "last_used": self.last_used.isoformat() if self.last_used else None,
            "is_enabled": self.is_enabled
        }
    
    def _record_usage(self):
        """Record that this tool was used"""
        self.usage_count += 1
        self.last_used = datetime.now()
    
    def run(self, query: str, **kwargs) -> ToolResult:
        """
        Public interface for running the tool
        
        This method handles common functionality like usage tracking,
        parameter validation, and error handling.
        
        Args:
            query: The input query
            **kwargs: Additional parameters
            
        Returns:
            ToolResult: The result of tool execution
        """
        if not self.is_enabled:
            return ToolResult(
                success=False,
                content="",
                metadata_json ={"tool_name": self.name},
                error_message=f"Tool '{self.name}' is currently disabled"
            )
        
        # Validate parameters
        if not self.validate_parameters(**kwargs):
            return ToolResult(
                success=False,
                content="",
                metadata_json ={"tool_name": self.name},
                error_message="Invalid parameters provided to tool"
            )
        
        # Record usage and execute
        start_time = datetime.now()
        
        try:
            self._record_usage()
            result = self.execute(query, **kwargs)
            
            # Add execution time to result
            execution_time = (datetime.now() - start_time).total_seconds()
            result.execution_time = execution_time
            
            # Ensure metadata_json includes tool name
            if "tool_name" not in result.metadata_json:
                result.metadata_json["tool_name"] = self.name
            
            return result
            
        except Exception as e:
            execution_time = (datetime.now() - start_time).total_seconds()
            
            return ToolResult(
                success=False,
                content="",
                metadata_json ={"tool_name": self.name},
                error_message=f"Tool execution failed: {str(e)}",
                execution_time=execution_time
            )
    
    def __str__(self) -> str:
        return f"{self.name}: {self.description}"
    
    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}(name='{self.name}')>"

Tool Manager

Create src/tools/tool_manager.py:

"""
Tool Manager for AI Agent

This module manages the collection of tools available to the agent,
handles tool selection, and provides a unified interface for tool execution.
"""

from typing import Dict, List, Optional, Any
import logging
from .base_tool import BaseTool, ToolResult

class ToolManager:
    """
    Manages tools for the AI agent
    
    The ToolManager acts as a registry and coordinator for all tools.
    It handles tool registration, selection, and execution.
    """
    
    def __init__(self):
        self.tools: Dict[str, BaseTool] = {}
        self.logger = logging.getLogger(__name__)
        
    def register_tool(self, tool: BaseTool) -> None:
        """
        Register a tool with the manager
        
        Args:
            tool: The tool instance to register
        """
        if tool.name in self.tools:
            self.logger.warning(f"Tool '{tool.name}' is already registered. Overwriting.")
        
        self.tools[tool.name] = tool
        self.logger.info(f"Registered tool: {tool.name}")
    
    def unregister_tool(self, tool_name: str) -> bool:
        """
        Unregister a tool
        
        Args:
            tool_name: Name of the tool to unregister
            
        Returns:
            True if tool was found and removed, False otherwise
        """
        if tool_name in self.tools:
            del self.tools[tool_name]
            self.logger.info(f"Unregistered tool: {tool_name}")
            return True
        return False
    
    def get_tool(self, tool_name: str) -> Optional[BaseTool]:
        """
        Get a tool by name
        
        Args:
            tool_name: Name of the tool to retrieve
            
        Returns:
            The tool instance or None if not found
        """
        return self.tools.get(tool_name)
    
    def list_tools(self) -> List[Dict[str, Any]]:
        """
        Get a list of all registered tools with their information
        
        Returns:
            List of dictionaries containing tool information
        """
        return [
            {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.get_parameters(),
                "usage_stats": tool.get_usage_stats()
            }
            for tool in self.tools.values()
        ]
    
    def execute_tool(self, tool_name: str, query: str, **kwargs) -> ToolResult:
        """
        Execute a specific tool
        
        Args:
            tool_name: Name of the tool to execute
            query: Query to pass to the tool
            **kwargs: Additional parameters for the tool
            
        Returns:
            ToolResult: Result of tool execution
        """
        tool = self.get_tool(tool_name)
        
        if tool is None:
            return ToolResult(
                success=False,
                content="",
                metadata_json ={"requested_tool": tool_name},
                error_message=f"Tool '{tool_name}' not found"
            )
        
        self.logger.debug(f"Executing tool '{tool_name}' with query: {query}")
        
        result = tool.run(query, **kwargs)
        
        self.logger.debug(f"Tool '{tool_name}' execution completed. Success: {result.success}")
        
        return result
    
    def get_tool_suggestions(self, query: str) -> List[str]:
        """
        Suggest tools that might be relevant for a given query
        
        This is a simple implementation that can be enhanced with
        more sophisticated matching algorithms.
        
        Args:
            query: The user's query
            
        Returns:
            List of tool names that might be relevant
        """
        query_lower = query.lower()
        suggestions = []
        
        # Simple keyword matching - can be enhanced
        for tool_name, tool in self.tools.items():
            if not tool.is_enabled:
                continue
                
            # Check if query contains keywords related to the tool
            tool_keywords = tool.description.lower().split()
            
            if any(keyword in query_lower for keyword in tool_keywords):
                suggestions.append(tool_name)
        
        return suggestions
    
    def get_manager_stats(self) -> Dict[str, Any]:
        """Get statistics about the tool manager"""
        total_tools = len(self.tools)
        enabled_tools = sum(1 for tool in self.tools.values() if tool.is_enabled)
        total_usage = sum(tool.usage_count for tool in self.tools.values())
        
        return {
            "total_tools": total_tools,
            "enabled_tools": enabled_tools,
            "disabled_tools": total_tools - enabled_tools,
            "total_tool_usage": total_usage,
            "tools": {name: tool.get_usage_stats() for name, tool in self.tools.items()}
        }

---

Step 3: Implementing Web Search Tool

We’ll start with web search so the agent can pull in fresh, real‑time information.

Now let's create our first concrete tool - a web search capability that gives your agent access to real-time information.

Web Search Tool Implementation

Create src/tools/web_search_tool.py:

"""
Web Search Tool for AI Agent

This tool provides web search capabilities using DuckDuckGo's Instant Answer API
and web scraping for more comprehensive results.
"""

import requests
import time
from typing import Dict, Any, List, Optional
from urllib.parse import quote_plus
import json
from bs4 import BeautifulSoup
import re

from .base_tool import BaseTool, ToolResult

class WebSearchTool(BaseTool):
    """
    Web search tool using multiple search strategies
    
    This tool can search the web using different approaches:
    1. DuckDuckGo Instant Answer API for quick facts
    2. Web scraping for more detailed results
    3. Multiple search engines for comprehensive coverage
    """
    
    def __init__(self):
        super().__init__(
            name="web_search",
            description="Search the web for current information, news, facts, and answers to questions"
        )
        
        # Configuration
        self.timeout = 10
        self.max_results = 5
        self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        
        # Rate limiting
        self.last_request_time = 0
        self.min_request_interval = 1.0  # Minimum seconds between requests
    
    def get_parameters(self) -> Dict[str, Any]:
        """Get parameters this tool accepts"""
        return {
            "query": {
                "type": "string",
                "description": "The search query",
                "required": True
            },
            "max_results": {
                "type": "integer",
                "description": "Maximum number of results to return",
                "required": False,
                "default": 5
            },
            "search_type": {
                "type": "string",
                "description": "Type of search: 'instant', 'web', or 'comprehensive'",
                "required": False,
                "default": "comprehensive"
            }
        }
    
    def validate_parameters(self, **kwargs) -> bool:
        """Validate search parameters"""
        query = kwargs.get('query', '')
        
        if not query or not query.strip():
            return False
        
        max_results = kwargs.get('max_results', 5)
        if not isinstance(max_results, int) or max_results < 1 or max_results > 20:
            return False
        
        search_type = kwargs.get('search_type', 'comprehensive')
        if search_type not in ['instant', 'web', 'comprehensive']:
            return False
        
        return True
    
    def execute(self, query: str, **kwargs) -> ToolResult:
        """
        Execute web search
        
        Args:
            query: Search query
            **kwargs: Additional parameters (max_results, search_type)
            
        Returns:
            ToolResult with search results
        """
        max_results = kwargs.get('max_results', self.max_results)
        search_type = kwargs.get('search_type', 'comprehensive')
        
        # Rate limiting
        self._enforce_rate_limit()
        
        try:
            if search_type == 'instant':
                return self._instant_search(query)
            elif search_type == 'web':
                return self._web_search(query, max_results)
            else:  # comprehensive
                return self._comprehensive_search(query, max_results)
                
        except Exception as e:
            return ToolResult(
                success=False,
                content="",
                metadata_json ={
                    "query": query,
                    "search_type": search_type,
                    "error_type": type(e).__name__
                },
                error_message=f"Search failed: {str(e)}"
            )
    
    def _enforce_rate_limit(self):
        """Enforce rate limiting between requests"""
        current_time = time.time()
        time_since_last = current_time - self.last_request_time
        
        if time_since_last < self.min_request_interval:
            sleep_time = self.min_request_interval - time_since_last
            time.sleep(sleep_time)
        
        self.last_request_time = time.time()
    
    def _instant_search(self, query: str) -> ToolResult:
        """
        Perform instant search using DuckDuckGo Instant Answer API
        
        Args:
            query: Search query
            
        Returns:
            ToolResult with instant answer
        """
        url = "https://api.duckduckgo.com/"
        params = {
            'q': query,
            'format': 'json',
            'no_html': '1',
            'skip_disambig': '1'
        }
        
        response = requests.get(url, params=params, timeout=self.timeout)
        response.raise_for_status()
        
        data = response.json()
        
        # Extract the most relevant information
        content_parts = []
        
        # Abstract (main answer)
        if data.get('Abstract'):
            content_parts.append(f"**Answer:** {data['Abstract']}")
        
        # Definition
        if data.get('Definition'):
            content_parts.append(f"**Definition:** {data['Definition']}")
        
        # Answer (direct answer)
        if data.get('Answer'):
            content_parts.append(f"**Direct Answer:** {data['Answer']}")
        
        # Related topics
        if data.get('RelatedTopics'):
            topics = [topic.get('Text', '') for topic in data['RelatedTopics'][:3] if topic.get('Text')]
            if topics:
                content_parts.append(f"**Related:** {'; '.join(topics)}")
        
        if content_parts:
            content = '\n\n'.join(content_parts)
        else:
            content = "No instant answer found. Try a web search for more comprehensive results."
        
        return ToolResult(
            success=True,
            content=content,
            metadata_json ={
                "query": query,
                "search_type": "instant",
                "source": "DuckDuckGo Instant Answer",
                "has_abstract": bool(data.get('Abstract')),
                "has_definition": bool(data.get('Definition')),
                "related_topics_count": len(data.get('RelatedTopics', []))
            }
        )
    
    def _web_search(self, query: str, max_results: int) -> ToolResult:
        """
        Perform web search using search engine scraping
        
        Args:
            query: Search query
            max_results: Maximum number of results
            
        Returns:
            ToolResult with search results
        """
        # Use DuckDuckGo HTML search (more results than instant API)
        search_url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
        
        headers = {
            'User-Agent': self.user_agent
        }
        
        response = requests.get(search_url, headers=headers, timeout=self.timeout)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.content, 'html.parser')
        
        # Extract search results
        results = []
        result_elements = soup.find_all('div', class_='result')
        
        for element in result_elements[:max_results]:
            try:
                # Extract title
                title_elem = element.find('a', class_='result__a')
                title = title_elem.get_text(strip=True) if title_elem else "No title"
                
                # Extract URL
                url = title_elem.get('href') if title_elem else ""
                
                # Extract snippet
                snippet_elem = element.find('a', class_='result__snippet')
                snippet = snippet_elem.get_text(strip=True) if snippet_elem else ""
                
                if title and snippet:
                    results.append({
                        'title': title,
                        'url': url,
                        'snippet': snippet
                    })
                    
            except Exception as e:
                # Skip malformed results
                continue
        
        if not results:
            return ToolResult(
                success=False,
                content="No search results found.",
                metadata_json ={
                    "query": query,
                    "search_type": "web",
                    "results_count": 0
                },
                error_message="No search results could be extracted"
            )
        
        # Format results
        content_parts = [f"**Search Results for:** {query}\n"]
        
        for i, result in enumerate(results, 1):
            content_parts.append(f"**{i}. {result['title']}**")
            content_parts.append(f"{result['snippet']}")
            if result['url']:
                content_parts.append(f"*Source: {result['url']}*")
            content_parts.append("")  # Empty line between results
        
        content = '\n'.join(content_parts)
        
        return ToolResult(
            success=True,
            content=content,
            metadata_json ={
                "query": query,
                "search_type": "web",
                "results_count": len(results),
                "source": "DuckDuckGo Web Search"
            }
        )
    
    def _comprehensive_search(self, query: str, max_results: int) -> ToolResult:
        """
        Perform comprehensive search combining instant and web results
        
        Args:
            query: Search query
            max_results: Maximum number of results
            
        Returns:
            ToolResult with comprehensive search results
        """
        results = []
        
        # Try instant search first
        try:
            instant_result = self._instant_search(query)
            if instant_result.success and instant_result.content.strip():
                # Only include if we got meaningful content
                if "No instant answer found" not in instant_result.content:
                    results.append(("Instant Answer", instant_result.content))
        except Exception:
            # Continue if instant search fails
            pass
        
        # Add web search results
        try:
            web_result = self._web_search(query, max_results)
            if web_result.success:
                results.append(("Web Search Results", web_result.content))
        except Exception as e:
            # If web search also fails, return error
            if not results:  # No instant results either
                return ToolResult(
                    success=False,
                    content="",
                    metadata_json ={
                        "query": query,
                        "search_type": "comprehensive"
                    },
                    error_message=f"All search methods failed: {str(e)}"
                )
        
        if not results:
            return ToolResult(
                success=False,
                content="No search results found from any source.",
                metadata_json ={
                    "query": query,
                    "search_type": "comprehensive"
                },
                error_message="No results from instant or web search"
            )
        
        # Combine results
        content_parts = []
        for section_title, section_content in results:
            content_parts.append(f"## {section_title}")
            content_parts.append(section_content)
            content_parts.append("")  # Empty line between sections
        
        content = '\n'.join(content_parts)
        
        return ToolResult(
            success=True,
            content=content,
            metadata_json ={
                "query": query,
                "search_type": "comprehensive",
                "sections_count": len(results),
                "has_instant_answer": any("Instant Answer" in title for title, _ in results),
                "has_web_results": any("Web Search" in title for title, _ in results)
            }
        )

---

Step 4: Creating Additional API Integration Tools

Let's add more tools to demonstrate different types of API integrations.

Weather API Tool

Create src/tools/weather_tool.py:

"""
Weather Tool for AI Agent

This tool provides weather information using a free weather API.
"""

import requests
from typing import Dict, Any
from .base_tool import BaseTool, ToolResult

class WeatherTool(BaseTool):
    """
    Weather information tool using OpenWeatherMap API
    
    This tool can get current weather, forecasts, and weather alerts
    for any location worldwide.
    """
    
    def __init__(self, api_key: str = None):
        super().__init__(
            name="weather",
            description="Get current weather information and forecasts for any location"
        )
        
        self.api_key = api_key
        self.base_url = "http://api.openweathermap.org/data/2.5"
        self.timeout = 10
    
    def get_parameters(self) -> Dict[str, Any]:
        """Get parameters this tool accepts"""
        return {
            "location": {
                "type": "string",
                "description": "City name, state/country (e.g., 'New York, NY' or 'London, UK')",
                "required": True
            },
            "units": {
                "type": "string",
                "description": "Temperature units: 'metric', 'imperial', or 'kelvin'",
                "required": False,
                "default": "metric"
            },
            "forecast_days": {
                "type": "integer",
                "description": "Number of forecast days (0 for current weather only)",
                "required": False,
                "default": 0
            }
        }
    
    def validate_parameters(self, **kwargs) -> bool:
        """Validate weather parameters"""
        location = kwargs.get('location', '')
        if not location or not location.strip():
            return False
        
        units = kwargs.get('units', 'metric')
        if units not in ['metric', 'imperial', 'kelvin']:
            return False
        
        forecast_days = kwargs.get('forecast_days', 0)
        if not isinstance(forecast_days, int) or forecast_days < 0 or forecast_days > 5:
            return False
        
        return True
    
    def execute(self, query: str, **kwargs) -> ToolResult:
        """
        Execute weather lookup
        
        Args:
            query: Location query (can be used instead of location parameter)
            **kwargs: Additional parameters (location, units, forecast_days)
            
        Returns:
            ToolResult with weather information
        """
        if not self.api_key:
            return ToolResult(
                success=False,
                content="",
                metadata_json ={"tool_name": self.name},
                error_message="Weather API key not configured"
            )
        
        # Use location parameter or fall back to query
        location = kwargs.get('location', query)
        units = kwargs.get('units', 'metric')
        forecast_days = kwargs.get('forecast_days', 0)
        
        try:
            if forecast_days == 0:
                return self._get_current_weather(location, units)
            else:
                return self._get_weather_forecast(location, units, forecast_days)
                
        except Exception as e:
            return ToolResult(
                success=False,
                content="",
                metadata_json ={
                    "location": location,
                    "units": units,
                    "error_type": type(e).__name__
                },
                error_message=f"Weather lookup failed: {str(e)}"
            )
    
    def _get_current_weather(self, location: str, units: str) -> ToolResult:
        """Get current weather for location"""
        url = f"{self.base_url}/weather"
        params = {
            'q': location,
            'appid': self.api_key,
            'units': units
        }
        
        response = requests.get(url, params=params, timeout=self.timeout)
        response.raise_for_status()
        
        data = response.json()
        
        # Extract weather information
        temp = data['main']['temp']
        feels_like = data['main']['feels_like']
        humidity = data['main']['humidity']
        description = data['weather'][0]['description'].title()
        city = data['name']
        country = data['sys']['country']
        
        # Format temperature unit
        unit_symbol = {'metric': '°C', 'imperial': '°F', 'kelvin': 'K'}[units]
        
        content = f"""**Current Weather for {city}, {country}**

🌡️ **Temperature:** {temp}{unit_symbol} (feels like {feels_like}{unit_symbol})
🌤️ **Conditions:** {description}
💧 **Humidity:** {humidity}%
"""
        
        return ToolResult(
            success=True,
            content=content,
            metadata_json ={
                "location": f"{city}, {country}",
                "temperature": temp,
                "units": units,
                "conditions": description,
                "humidity": humidity,
                "source": "OpenWeatherMap"
            }
        )
    
    def _get_weather_forecast(self, location: str, units: str, days: int) -> ToolResult:
        """Get weather forecast for location"""
        url = f"{self.base_url}/forecast"
        params = {
            'q': location,
            'appid': self.api_key,
            'units': units
        }
        
        response = requests.get(url, params=params, timeout=self.timeout)
        response.raise_for_status()
        
        data = response.json()
        
        city = data['city']['name']
        country = data['city']['country']
        
        # Format temperature unit
        unit_symbol = {'metric': '°C', 'imperial': '°F', 'kelvin': 'K'}[units]
        
        content_parts = [f"**{days}-Day Weather Forecast for {city}, {country}**\n"]
        
        # Process forecast data (API returns 3-hour intervals)
        forecasts = data['list'][:days * 8]  # Approximate daily forecasts
        
        current_date = None
        for forecast in forecasts[::8]:  # Take every 8th entry (roughly daily)
            date = forecast['dt_txt'].split(' ')[0]
            temp = forecast['main']['temp']
            description = forecast['weather'][0]['description'].title()
            
            content_parts.append(f"📅 **{date}:** {temp}{unit_symbol}, {description}")
        
        content = '\n'.join(content_parts)
        
        return ToolResult(
            success=True,
            content=content,
            metadata_json ={
                "location": f"{city}, {country}",
                "forecast_days": days,
                "units": units,
                "source": "OpenWeatherMap"
            }
        )

Calculator Tool

Create src/tools/calculator_tool.py:

"""
Calculator Tool for AI Agent

This tool provides mathematical calculation capabilities.
"""

import re
import math
from typing import Dict, Any
from .base_tool import BaseTool, ToolResult

class CalculatorTool(BaseTool):
    """
    Mathematical calculator tool
    
    This tool can perform various mathematical operations including:
    - Basic arithmetic (+, -, *, /)
    - Advanced functions (sin, cos, tan, log, sqrt, etc.)
    - Constants (pi, e)
    """
    
    def __init__(self):
        super().__init__(
            name="calculator",
            description="Perform mathematical calculations and solve equations"
        )
        
        # Safe mathematical functions
        self.safe_functions = {
            'sin': math.sin,
            'cos': math.cos,
            'tan': math.tan,
            'asin': math.asin,
            'acos': math.acos,
            'atan': math.atan,
            'sinh': math.sinh,
            'cosh': math.cosh,
            'tanh': math.tanh,
            'log': math.log,
            'log10': math.log10,
            'log2': math.log2,
            'sqrt': math.sqrt,
            'abs': abs,
            'ceil': math.ceil,
            'floor': math.floor,
            'round': round,
            'pow': pow,
            'exp': math.exp,
            'pi': math.pi,
            'e': math.e,
        }
    
    def get_parameters(self) -> Dict[str, Any]:
        """Get parameters this tool accepts"""
        return {
            "expression": {
                "type": "string",
                "description": "Mathematical expression to evaluate (e.g., '2 + 3 * 4', 'sin(pi/2)', 'sqrt(16)')",
                "required": True
            },
            "precision": {
                "type": "integer",
                "description": "Number of decimal places for the result",
                "required": False,
                "default": 6
            }
        }
    
    def validate_parameters(self, **kwargs) -> bool:
        """Validate calculator parameters"""
        expression = kwargs.get('expression', '')
        if not expression or not expression.strip():
            return False
        
        precision = kwargs.get('precision', 6)
        if not isinstance(precision, int) or precision < 0 or precision > 15:
            return False
        
        return True
    
    def execute(self, query: str, **kwargs) -> ToolResult:
        """
        Execute mathematical calculation
        
        Args:
            query: Mathematical expression (can be used instead of expression parameter)
            **kwargs: Additional parameters (expression, precision)
            
        Returns:
            ToolResult with calculation result
        """
        # Use expression parameter or fall back to query
        expression = kwargs.get('expression', query)
        precision = kwargs.get('precision', 6)
        
        try:
            # Clean and validate the expression
            cleaned_expression = self._clean_expression(expression)
            
            if not self._is_safe_expression(cleaned_expression):
                return ToolResult(
                    success=False,
                    content="",
                    metadata_json ={"expression": expression},
                    error_message="Expression contains unsafe operations"
                )
            
            # Evaluate the expression
            result = self._evaluate_expression(cleaned_expression)
            
            # Format the result
            if isinstance(result, float):
                if result.is_integer():
                    formatted_result = str(int(result))
                else:
                    formatted_result = f"{result:.{precision}f}".rstrip('0').rstrip('.')
            else:
                formatted_result = str(result)
            
            content = f"**Calculation:** {expression}\n**Result:** {formatted_result}"
            
            return ToolResult(
                success=True,
                content=content,
                metadata_json ={
                    "expression": expression,
                    "result": result,
                    "formatted_result": formatted_result,
                    "precision": precision
                }
            )
            
        except Exception as e:
            return ToolResult(
                success=False,
                content="",
                metadata_json ={
                    "expression": expression,
                    "error_type": type(e).__name__
                },
                error_message=f"Calculation failed: {str(e)}"
            )
    
    def _clean_expression(self, expression: str) -> str:
        """Clean and normalize the mathematical expression"""
        # Remove whitespace
        cleaned = re.sub(r'\s+', '', expression)
        
        # Replace common mathematical notation
        replacements = {
            '^': '**',  # Power operator
            'π': 'pi',  # Pi constant
            '×': '*',   # Multiplication
            '÷': '/',   # Division
        }
        
        for old, new in replacements.items():
            cleaned = cleaned.replace(old, new)
        
        return cleaned
    
    def _is_safe_expression(self, expression: str) -> bool:
        """Check if the expression is safe to evaluate"""
        # List of dangerous patterns
        dangerous_patterns = [
            r'__',          # Double underscore (Python internals)
            r'import',      # Import statements
            r'exec',        # Code execution
            r'eval',        # Code evaluation
            r'open',        # File operations
            r'file',        # File operations
            r'input',       # User input
            r'raw_input',   # User input
            r'globals',     # Global variables
            r'locals',      # Local variables
            r'vars',        # Variable inspection
            r'dir',         # Directory listing
            r'help',        # Help system
            r'quit',        # Exit functions
            r'exit',        # Exit functions
        ]
        
        for pattern in dangerous_patterns:
            if re.search(pattern, expression, re.IGNORECASE):
                return False
        
        return True
    
    def _evaluate_expression(self, expression: str):
        """Safely evaluate the mathematical expression"""
        # Create a safe namespace with only mathematical functions
        safe_dict = {
            "__builtins__": {},
            **self.safe_functions
        }
        
        # Evaluate the expression
        result = eval(expression, safe_dict)
        
        return result

---

Step 5: Integrating Tools with Your Agent

Now let's update your agent to use the tool system.

Enhanced Agent with Tool Support

Create src/agents/tool_agent.py:

"""
Tool-enabled AI Agent

This agent extends the base agent with tool usage capabilities.
"""

from typing import List, Dict, Any, Optional
import re
import json

from .memory_agent import MemoryAgent
from ..tools.tool_manager import ToolManager
from ..tools.web_search_tool import WebSearchTool
from ..tools.weather_tool import WeatherTool
from ..tools.calculator_tool import CalculatorTool
from ..utils.config import AgentConfig

class ToolAgent(MemoryAgent):
    """
    AI Agent with tool usage capabilities
    
    This agent can use various tools to perform actions and retrieve information
    beyond what's possible with just conversation.
    """
    
    def __init__(self, config: AgentConfig, database_url: str = None):
        super().__init__(config, database_url)
        
        # Initialize tool manager
        self.tool_manager = ToolManager()
        
        # Register default tools
        self._register_default_tools()
        
        # Enhanced system prompt that includes tool usage
        self.tool_system_prompt = self._create_tool_system_prompt()
    
    def _register_default_tools(self):
        """Register default tools with the agent"""
        
        # Web search tool
        web_search = WebSearchTool()
        self.tool_manager.register_tool(web_search)
        
        # Calculator tool
        calculator = CalculatorTool()
        self.tool_manager.register_tool(calculator)
        
        # Weather tool (if API key is available)
        weather_api_key = getattr(self.config, 'weather_api_key', None)
        if weather_api_key:
            weather = WeatherTool(weather_api_key)
            self.tool_manager.register_tool(weather)
        
        self.logger.info(f"Registered {len(self.tool_manager.tools)} tools")
    
    def _create_tool_system_prompt(self) -> str:
        """Create system prompt that includes tool usage instructions"""
        
        tools_info = []
        for tool_info in self.tool_manager.list_tools():
            tools_info.append(f"- **{tool_info['name']}**: {tool_info['description']}")
        
        tools_list = '\n'.join(tools_info)
        
        return f"""{self.config.system_prompt}

You have access to the following tools that can help you provide better assistance:

{tools_list}

When you need to use a tool, respond with a tool usage request in this format:
[TOOL: tool_name] query or parameters

For example:
- [TOOL: web_search] latest news about AI agents
- [TOOL: calculator] 15 * 24 + 36
- [TOOL: weather] New York, NY

After using a tool, incorporate the results naturally into your response to the user.
"""
    
    def chat(self, message: str, conversation_id: Optional[str] = None) -> str:
        """
        Enhanced chat method with tool usage
        
        Args:
            message: User message
            conversation_id: Optional conversation ID
            
        Returns:
            Agent's response, potentially enhanced with tool results
        """
        if not self.is_initialized:
            self.initialize()
        
        # Create or get conversation
        if conversation_id is None:
            conversation_id = self.create_conversation()
        
        context = self.get_conversation(conversation_id)
        if context is None:
            raise ValueError(f"Conversation {conversation_id} not found")
        
        # Add user message
        context.add_message("user", message)
        
        try:
            # Get initial response from AI
            initial_response = self._get_completion_with_memory(context, self.tool_system_prompt)
            
            # Check if the response contains tool usage requests
            enhanced_response = self._process_tool_requests(initial_response, conversation_id)
            
            # Add final response to conversation
            context.add_message("assistant", enhanced_response)
            
            self.logger.info(f"Conversation {conversation_id}: User -> Agent (with tools)")
            return enhanced_response
            
        except Exception as e:
            self.logger.error(f"Error in tool-enabled chat: {e}")
            error_response = "I apologize, but I encountered an error processing your request. Please try again."
            context.add_message("assistant", error_response)
            return error_response
    
    def _process_tool_requests(self, response: str, conversation_id: str) -> str:
        """
        Process tool usage requests in the AI's response
        
        Args:
            response: Initial AI response that may contain tool requests
            conversation_id: Current conversation ID
            
        Returns:
            Enhanced response with tool results
        """
        # Pattern to match tool usage requests
        tool_pattern = r'\[TOOL:\s*(\w+)\]\s*(.+?)(?=\[TOOL:|$)'
        
        tool_matches = re.findall(tool_pattern, response, re.DOTALL | re.IGNORECASE)
        
        if not tool_matches:
            # No tool usage requested
            return response
        
        # Process each tool request
        tool_results = []
        
        for tool_name, query in tool_matches:
            tool_name = tool_name.strip()
            query = query.strip()
            
            self.logger.debug(f"Processing tool request: {tool_name} with query: {query}")
            
            # Execute the tool
            result = self.tool_manager.execute_tool(tool_name, query)
            
            if result.success:
                tool_results.append({
                    'tool': tool_name,
                    'query': query,
                    'result': result.content,
                    'success': True
                })
                self.logger.debug(f"Tool {tool_name} executed successfully")
            else:
                tool_results.append({
                    'tool': tool_name,
                    'query': query,
                    'error': result.error_message,
                    'success': False
                })
                self.logger.warning(f"Tool {tool_name} failed: {result.error_message}")
        
        # Generate enhanced response with tool results
        enhanced_response = self._generate_enhanced_response(response, tool_results, conversation_id)
        
        return enhanced_response
    
    def _generate_enhanced_response(self, original_response: str, tool_results: List[Dict], conversation_id: str) -> str:
        """
        Generate enhanced response incorporating tool results
        
        Args:
            original_response: Original AI response with tool requests
            tool_results: Results from tool executions
            conversation_id: Current conversation ID
            
        Returns:
            Enhanced response with tool results integrated
        """
        # Create a summary of tool results
        tool_summary = []
        
        for result in tool_results:
            if result['success']:
                tool_summary.append(f"**{result['tool'].title()} Results:**\n{result['result']}")
            else:
                tool_summary.append(f"**{result['tool'].title()} Error:** {result['error']}")
        
        # Combine tool results
        tools_content = '\n\n'.join(tool_summary)
        
        # Create context for generating final response
        context = self.get_conversation(conversation_id)
        
        # Add tool results as a system message for context
        enhancement_prompt = f"""Based on the following tool results, provide a comprehensive and helpful response to the user:

{tools_content}

Integrate this information naturally into your response. Don't just repeat the tool results - synthesize them into a coherent, helpful answer."""
        
        # Get enhanced response from AI
        try:
            # Temporarily add enhancement context
            context.add_message("system", enhancement_prompt)
            
            enhanced_response = self._get_completion_with_memory(context)
            
            # Remove the temporary system message
            context.messages.pop()  # Remove the enhancement prompt
            
            return enhanced_response
            
        except Exception as e:
            self.logger.error(f"Error generating enhanced response: {e}")
            
            # Fallback: return original response with tool results appended
            return f"{original_response}\n\n{tools_content}"
    
    def list_available_tools(self) -> List[Dict[str, Any]]:
        """Get list of available tools"""
        return self.tool_manager.list_tools()
    
    def get_tool_stats(self) -> Dict[str, Any]:
        """Get tool usage statistics"""
        return self.tool_manager.get_manager_stats()
    
    def register_tool(self, tool):
        """Register a new tool with the agent"""
        self.tool_manager.register_tool(tool)
        
        # Update system prompt to include new tool
        self.tool_system_prompt = self._create_tool_system_prompt()
        
        self.logger.info(f"Registered new tool: {tool.name}")
    
    def unregister_tool(self, tool_name: str) -> bool:
        """Unregister a tool from the agent"""
        success = self.tool_manager.unregister_tool(tool_name)
        
        if success:
            # Update system prompt
            self.tool_system_prompt = self._create_tool_system_prompt()
            self.logger.info(f"Unregistered tool: {tool_name}")
        
        return success

---

Step 6: Testing Your Tool-Enabled Agent

Let's create comprehensive tests for the tool functionality.

Tool Tests

Create tests/test_tools.py:

"""
Tests for AI Agent tools
"""

import pytest
from unittest.mock import Mock, patch, MagicMock
import requests

from src.tools.web_search_tool import WebSearchTool
from src.tools.calculator_tool import CalculatorTool
from src.tools.weather_tool import WeatherTool
from src.tools.tool_manager import ToolManager

class TestWebSearchTool:
    """Test web search tool functionality"""
    
    def test_tool_initialization(self):
        """Test web search tool initializes correctly"""
        tool = WebSearchTool()
        
        assert tool.name == "web_search"
        assert "search" in tool.description.lower()
        assert tool.is_enabled
        assert tool.usage_count == 0
    
    def test_parameter_validation(self):
        """Test parameter validation"""
        tool = WebSearchTool()
        
        # Valid parameters
        assert tool.validate_parameters(query="test query")
        assert tool.validate_parameters(query="test", max_results=5, search_type="web")
        
        # Invalid parameters
        assert not tool.validate_parameters(query="")
        assert not tool.validate_parameters(query="test", max_results=0)
        assert not tool.validate_parameters(query="test", search_type="invalid")
    
    @patch('src.tools.web_search_tool.requests.get')
    def test_instant_search(self, mock_get):
        """Test instant search functionality"""
        # Mock API response
        mock_response = Mock()
        mock_response.json.return_value = {
            'Abstract': 'Test abstract answer',
            'Definition': 'Test definition',
            'RelatedTopics': [{'Text': 'Related topic 1'}]
        }
        mock_response.raise_for_status.return_value = None
        mock_get.return_value = mock_response
        
        tool = WebSearchTool()
        result = tool.execute("test query", search_type="instant")
        
        assert result.success
        assert "Test abstract answer" in result.content
        assert result.metadata_json["search_type"] == "instant"
    
    @patch('src.tools.web_search_tool.requests.get')
    def test_search_error_handling(self, mock_get):
        """Test search error handling"""
        # Mock API error
        mock_get.side_effect = requests.RequestException("API Error")
        
        tool = WebSearchTool()
        result = tool.execute("test query")
        
        assert not result.success
        assert "failed" in result.error_message.lower()

class TestCalculatorTool:
    """Test calculator tool functionality"""
    
    def test_tool_initialization(self):
        """Test calculator tool initializes correctly"""
        tool = CalculatorTool()
        
        assert tool.name == "calculator"
        assert "mathematical" in tool.description.lower()
        assert tool.is_enabled
    
    def test_basic_arithmetic(self):
        """Test basic arithmetic operations"""
        tool = CalculatorTool()
        
        # Addition
        result = tool.execute("2 + 3")
        assert result.success
        assert "5" in result.content
        
        # Multiplication
        result = tool.execute("4 * 5")
        assert result.success
        assert "20" in result.content
        
        # Division
        result = tool.execute("10 / 2")
        assert result.success
        assert "5" in result.content
    
    def test_advanced_functions(self):
        """Test advanced mathematical functions"""
        tool = CalculatorTool()
        
        # Square root
        result = tool.execute("sqrt(16)")
        assert result.success
        assert "4" in result.content
        
        # Trigonometric functions
        result = tool.execute("sin(0)")
        assert result.success
        assert "0" in result.content
    
    def test_expression_safety(self):
        """Test that dangerous expressions are rejected"""
        tool = CalculatorTool()
        
        # Dangerous expressions should be rejected
        dangerous_expressions = [
            "__import__('os').system('ls')",
            "exec('print(1)')",
            "eval('1+1')",
            "open('/etc/passwd')"
        ]
        
        for expr in dangerous_expressions:
            result = tool.execute(expr)
            assert not result.success
            assert "unsafe" in result.error_message.lower()

class TestWeatherTool:
    """Test weather tool functionality"""
    
    def test_tool_initialization(self):
        """Test weather tool initializes correctly"""
        tool = WeatherTool("test-api-key")
        
        assert tool.name == "weather"
        assert "weather" in tool.description.lower()
        assert tool.api_key == "test-api-key"
    
    def test_no_api_key_error(self):
        """Test error when no API key is provided"""
        tool = WeatherTool()
        
        result = tool.execute("New York")
        assert not result.success
        assert "api key" in result.error_message.lower()
    
    @patch('src.tools.weather_tool.requests.get')
    def test_current_weather(self, mock_get):
        """Test current weather functionality"""
        # Mock API response
        mock_response = Mock()
        mock_response.json.return_value = {
            'main': {'temp': 20, 'feels_like': 22, 'humidity': 65},
            'weather': [{'description': 'clear sky'}],
            'name': 'New York',
            'sys': {'country': 'US'}
        }
        mock_response.raise_for_status.return_value = None
        mock_get.return_value = mock_response
        
        tool = WeatherTool("test-api-key")
        result = tool.execute("New York", forecast_days=0)
        
        assert result.success
        assert "New York" in result.content
        assert "20" in result.content
        assert "clear sky" in result.content.lower()

class TestToolManager:
    """Test tool manager functionality"""
    
    def test_tool_registration(self):
        """Test tool registration and management"""
        manager = ToolManager()
        tool = CalculatorTool()
        
        # Register tool
        manager.register_tool(tool)
        
        assert tool.name in manager.tools
        assert manager.get_tool(tool.name) == tool
        assert len(manager.list_tools()) == 1
    
    def test_tool_execution(self):
        """Test tool execution through manager"""
        manager = ToolManager()
        tool = CalculatorTool()
        manager.register_tool(tool)
        
        # Execute tool
        result = manager.execute_tool("calculator", "2 + 2")
        
        assert result.success
        assert "4" in result.content
    
    def test_tool_not_found(self):
        """Test handling of non-existent tools"""
        manager = ToolManager()
        
        result = manager.execute_tool("nonexistent", "test")
        
        assert not result.success
        assert "not found" in result.error_message.lower()
    
    def test_tool_suggestions(self):
        """Test tool suggestion functionality"""
        manager = ToolManager()
        
        # Register tools
        manager.register_tool(CalculatorTool())
        manager.register_tool(WebSearchTool())
        
        # Test suggestions
        calc_suggestions = manager.get_tool_suggestions("calculate 2 + 2")
        assert "calculator" in calc_suggestions
        
        search_suggestions = manager.get_tool_suggestions("search for information")
        assert "web_search" in search_suggestions

---

Step 7: Creating a CLI with Tool Support

Let's update the CLI to demonstrate tool usage.

Enhanced CLI

Create src/cli_with_tools.py:

"""
Enhanced CLI with tool support
"""

import sys
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.markdown import Markdown

from .agents.tool_agent import ToolAgent
from .utils.config import get_config
from .utils.logger import get_logger

class ToolAgentCLI:
    """Enhanced CLI for tool-enabled agent"""
    
    def __init__(self):
        self.console = Console()
        self.config = get_config()
        self.agent = ToolAgent(self.config)
        self.logger = get_logger()
        self.current_conversation_id = None
    
    def run(self):
        """Run the enhanced CLI"""
        try:
            # Initialize agent
            self.agent.initialize()
            
            # Display welcome
            self.display_welcome()
            
            # Main loop
            while True:
                try:
                    user_input = input("\n💬 You: ").strip()
                    
                    if not user_input:
                        continue
                    
                    if not self.process_command(user_input):
                        break
            except KeyboardInterrupt:
                self.console.print("\n👋 Exiting ToolAgent CLI. Goodbye!")
            except Exception as e:
                self.logger.error(f"CLI error: {e}")


    if __name__ == "__main__":
        cli = ToolAgentCLI()
        cli.run()

    ---

    ## Part 4 Summary & Next Steps

    Congratulations! In this part, you:
    - Designed a tool interface for your agent
    - Implemented a web search tool and integrated external APIs
    - Connected your agent to OpenAI and other services
    - Handled API errors and rate limits
    - Tested tool usage and integration logic

    Your agent can now interact with the outside world, retrieve data, and perform real-world tasks. This unlocks powerful new capabilities and sets the stage for production deployment.

    ### What’s Next?
    In [Part 5: Testing, Debugging, and Deployment](/posts/build-first-ai-agent-part-5-testing-deployment), you’ll learn how to rigorously test your agent, debug common issues, and deploy it to production environments.

    ---