Search the web using Brave Search API.
| 44 | |
| 45 | |
| 46 | class WebSearchTool(Tool): |
| 47 | """Search the web using Brave Search API.""" |
| 48 | |
| 49 | name = "web_search" |
| 50 | description = "Search the web. Returns titles, URLs, and snippets." |
| 51 | parameters = { |
| 52 | "type": "object", |
| 53 | "properties": { |
| 54 | "query": {"type": "string", "description": "Search query"}, |
| 55 | "count": { |
| 56 | "type": "integer", |
| 57 | "description": "Results (1-10)", |
| 58 | "minimum": 1, |
| 59 | "maximum": 10, |
| 60 | }, |
| 61 | }, |
| 62 | "required": ["query"], |
| 63 | } |
| 64 | |
| 65 | def __init__(self, api_key: str | None = None, max_results: int = 5): |
| 66 | self.api_key = api_key or os.environ.get("BRAVE_API_KEY", "") |
| 67 | self.max_results = max_results |
| 68 | |
| 69 | async def execute(self, query: str, count: int | None = None, **kwargs: Any) -> str: |
| 70 | if not self.api_key: |
| 71 | return "Error: BRAVE_API_KEY not configured" |
| 72 | |
| 73 | try: |
| 74 | n = min(max(count or self.max_results, 1), 10) |
| 75 | async with httpx.AsyncClient() as client: |
| 76 | r = await client.get( |
| 77 | "https://api.search.brave.com/res/v1/web/search", |
| 78 | params={"q": query, "count": n}, |
| 79 | headers={"Accept": "application/json", "X-Subscription-Token": self.api_key}, |
| 80 | timeout=10.0, |
| 81 | ) |
| 82 | r.raise_for_status() |
| 83 | |
| 84 | results = r.json().get("web", {}).get("results", []) |
| 85 | if not results: |
| 86 | return f"No results for: {query}" |
| 87 | |
| 88 | lines = [f"Results for: {query}\n"] |
| 89 | for i, item in enumerate(results[:n], 1): |
| 90 | lines.append(f"{i}. {item.get('title', '')}\n {item.get('url', '')}") |
| 91 | if desc := item.get("description"): |
| 92 | lines.append(f" {desc}") |
| 93 | return "\n".join(lines) |
| 94 | except Exception as e: |
| 95 | return f"Error: {e}" |
| 96 | |
| 97 | |
| 98 | class WebFetchTool(Tool): |
no outgoing calls
no test coverage detected