|
| 1 | +"""Serper Google Search Tool - https://serper.dev""" |
| 2 | + |
| 3 | +import os |
| 4 | +from typing import Any |
| 5 | + |
| 6 | +import httpx |
| 7 | + |
| 8 | +from easyagent.tool.manager import register_tool |
| 9 | + |
| 10 | + |
| 11 | +@register_tool |
| 12 | +class SerperSearch: |
| 13 | + """Google Search via Serper API""" |
| 14 | + |
| 15 | + name = "serper_search" |
| 16 | + type = "function" |
| 17 | + description = "Search the web using Google via Serper API. Returns top search results." |
| 18 | + parameters = { |
| 19 | + "type": "object", |
| 20 | + "properties": { |
| 21 | + "query": { |
| 22 | + "type": "string", |
| 23 | + "description": "The search query", |
| 24 | + }, |
| 25 | + "num_results": { |
| 26 | + "type": "integer", |
| 27 | + "description": "Number of results to return (default: 5, max: 10)", |
| 28 | + }, |
| 29 | + }, |
| 30 | + "required": ["query"], |
| 31 | + } |
| 32 | + |
| 33 | + _api_key: str | None = None |
| 34 | + |
| 35 | + def init(self) -> None: |
| 36 | + self._api_key = os.environ.get("SERPER_API_KEY") |
| 37 | + |
| 38 | + def execute(self, query: str, num_results: int = 5, **kwargs: Any) -> str: |
| 39 | + if not self._api_key: |
| 40 | + return "Error: SERPER_API_KEY environment variable not set" |
| 41 | + |
| 42 | + num_results = min(max(1, num_results), 10) |
| 43 | + |
| 44 | + try: |
| 45 | + response = httpx.post( |
| 46 | + "https://google.serper.dev/search", |
| 47 | + headers={ |
| 48 | + "X-API-KEY": self._api_key, |
| 49 | + "Content-Type": "application/json", |
| 50 | + }, |
| 51 | + json={"q": query, "num": num_results}, |
| 52 | + timeout=30, |
| 53 | + ) |
| 54 | + response.raise_for_status() |
| 55 | + data = response.json() |
| 56 | + return _format_results(data, num_results) |
| 57 | + except httpx.HTTPStatusError as e: |
| 58 | + return f"Error: HTTP {e.response.status_code} - {e.response.text}" |
| 59 | + except Exception as e: |
| 60 | + return f"Error: {e}" |
| 61 | + |
| 62 | + |
| 63 | +def _format_results(data: dict[str, Any], limit: int) -> str: |
| 64 | + """Format search results into readable text""" |
| 65 | + lines: list[str] = [] |
| 66 | + |
| 67 | + # Knowledge Graph (if present) |
| 68 | + if kg := data.get("knowledgeGraph"): |
| 69 | + lines.append(f"[Knowledge Graph] {kg.get('title', '')}") |
| 70 | + if desc := kg.get("description"): |
| 71 | + lines.append(f" {desc}") |
| 72 | + lines.append("") |
| 73 | + |
| 74 | + # Answer Box (if present) |
| 75 | + if answer := data.get("answerBox"): |
| 76 | + if snippet := answer.get("snippet") or answer.get("answer"): |
| 77 | + lines.append(f"[Answer] {snippet}") |
| 78 | + lines.append("") |
| 79 | + |
| 80 | + # Organic Results |
| 81 | + organic = data.get("organic", [])[:limit] |
| 82 | + for i, result in enumerate(organic, 1): |
| 83 | + title = result.get("title", "No title") |
| 84 | + link = result.get("link", "") |
| 85 | + snippet = result.get("snippet", "") |
| 86 | + lines.append(f"{i}. {title}") |
| 87 | + lines.append(f" URL: {link}") |
| 88 | + if snippet: |
| 89 | + lines.append(f" {snippet}") |
| 90 | + lines.append("") |
| 91 | + |
| 92 | + return "\n".join(lines).strip() or "No results found" |
0 commit comments