| import os | |
| from typing import Optional, Dict, List | |
| try: | |
| from tavily import TavilyClient | |
| TAVILY_AVAILABLE = True | |
| except ImportError: | |
| TavilyClient = None | |
| TAVILY_AVAILABLE = False | |
| class WebSearchService: | |
| def __init__(self): | |
| self.client = None | |
| api_key = os.getenv("TAVILY_API_KEY") | |
| if api_key and TAVILY_AVAILABLE: | |
| self.client = TavilyClient(api_key=api_key) | |
| def search(self, query: str, max_results: int = 5) -> Optional[List[Dict]]: | |
| """ | |
| Perform a web search using Tavily. | |
| Returns list of results with title, url, content. | |
| """ | |
| if not self.client: | |
| print("⚠️ Tavily API key not configured.") | |
| return [] | |
| try: | |
| response = self.client.search( | |
| query=query, | |
| max_results=max_results, | |
| include_answer=True, | |
| include_raw_content=False | |
| ) | |
| return response.get("results", []) | |
| except Exception as e: | |
| print(f"❌ Web search failed: {e}") | |
| return [] | |
| # Global instance | |
| web_search_service = WebSearchService() | |