tools/ai_tools.py

#!/usr/bin/env python3
"""
AI Tools for Lead Generation

Contains all AI tools used by the lead generation agent including:
- brave_search: Web search functionality
- send_email: Email sending functionality
"""

import os
import json
import requests
from strands import tool
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail

@tool
def brave_search(query: str, count: int = 10) -> str:
    """Search the web using Brave Search API for news and information via direct HTTP requests"""
    
    print(f"🔍 Searching Brave for: '{query}' (count: {count})")
    
    # Get Brave API key from environment variable
    brave_api_key = os.getenv("BRAVE_API_KEY")
    if not brave_api_key:
        raise ValueError("BRAVE_API_KEY environment variable is not set")
    
    try:
        # Brave Search API endpoint
        api_url = "https://api.search.brave.com/res/v1/web/search"
        
        # Set up headers with API key
        headers = {
            "Accept": "application/json",
            "Accept-Encoding": "gzip",
            "X-Subscription-Token": brave_api_key
        }
        
        # Set up query parameters
        params = {
            "q": query,
            "count": min(count, 20),  # API maximum is 20
            "offset": 0,
            "safesearch": "moderate",
            "freshness": "",
            "text_decorations": True,
            "spellcheck": True,
            "result_filter": "web,news"  # Include both web and news results
        }
        
        # Make the HTTP request
        print(f"📡 Making request to Brave Search API...")
        response = requests.get(api_url, headers=headers, params=params, timeout=30)
        
        # Check if request was successful
        if response.status_code != 200:
            error_msg = f"Brave Search API returned status {response.status_code}: {response.text}"
            print(f"❌ {error_msg}")
            return error_msg
            
        # Parse JSON response
        search_data = response.json()
        formatted_results = []
        
        # Process web results
        if "web" in search_data and "results" in search_data["web"]:
            web_results = search_data["web"]["results"]
            for idx, result in enumerate(web_results[:count], 1):
                formatted_result = f"""
Result {idx}:
Title: {result.get('title', 'N/A')}
URL: {result.get('url', 'N/A')}
Description: {result.get('description', 'N/A')}
---"""
                formatted_results.append(formatted_result)
        
        # Process news results if available
        if "news" in search_data and "results" in search_data["news"]:
            news_results = search_data["news"]["results"]
            if news_results:
                formatted_results.append("\n🗞️ NEWS RESULTS:")
                for idx, news in enumerate(news_results[:count], 1):
                    formatted_result = f"""
News {idx}:
Title: {news.get('title', 'N/A')}
URL: {news.get('url', 'N/A')}
Description: {news.get('description', 'N/A')}
Age: {news.get('age', 'N/A')}
---"""
                    formatted_results.append(formatted_result)
        
        final_result = "\n".join(formatted_results) if formatted_results else "No results found"
        print(f"✅ Found {len(formatted_results)} results from Brave Search API")
        return final_result
        
    except requests.exceptions.Timeout:
        error_msg = "Request to Brave Search API timed out"
        print(f"❌ {error_msg}")
        return error_msg
    except requests.exceptions.RequestException as e:
        error_msg = f"Network error with Brave Search API: {str(e)}"
        print(f"❌ {error_msg}")
        return error_msg
    except json.JSONDecodeError as e:
        error_msg = f"Failed to parse Brave Search API response: {str(e)}"
        print(f"❌ {error_msg}")
        return error_msg
    except Exception as e:
        error_msg = f"Error performing Brave search: {str(e)}"
        print(f"❌ {error_msg}")
        return error_msg

@tool
def send_email(to: str, html_content: str, subject: str) -> str:
    """Send an email to given address"""

    print(f"Sending to {to} content {html_content} and subject {subject}.")
    
    # Get SendGrid API key from environment variable
    sendgrid_api_key = os.getenv("SENDGRID_API_KEY")
    if not sendgrid_api_key:
        raise ValueError("SENDGRID_API_KEY environment variable is not set")
    
    message = Mail(
        from_email='kavukcu.tolga@gmail.com',
        to_emails=to,
        subject=subject,
        html_content=html_content
    )
    
    try:
        sg = SendGridAPIClient(sendgrid_api_key)
        # sg.set_sendgrid_data_residency("eu")
        # uncomment the above line if you are sending mail using a regional EU subuser
        response = sg.send(message)
        print(response.status_code)
        print(response.body)
        print(response.headers)
        return f"Mail is sent to {to}"
    except Exception as e:
        print(f"Error sending email: {str(e)}")
        return f"Failed to send email to {to}: {str(e)}"