The WebSearchTool enables mindX agents to search the web using Google Custom Search JSON API. It provides a robust interface for retrieving real-time information from the internet, with graceful fallback to mock results when API credentials are unavailable.
File: tools/web_search_tool.py
Class: WebSearchTool
Version: 1.0.0
Status: ✅ Active
httpx for async HTTP requestsclass WebSearchTool(BaseTool):
- api_key: Google Search API key
- search_engine_id: Custom Search Engine ID
- http_client: httpx.AsyncClient for HTTP requests
- config: Configuration object
GOOGLE_SEARCH_API_KEY: Google Custom Search API keyGOOGLE_SEARCH_ENGINE_ID: Custom Search Engine ID# In config file or environment
tools.web_search.google_api_key: "your_api_key"
tools.web_search.google_search_engine_id: "your_engine_id"
tools.web_search.timeout_seconds: 20.0
from tools.web_search_tool import WebSearchTool
from utils.config import Config
config = Config()
tool = WebSearchTool(config=config)
Execute search
results = await tool.execute(
query="latest advancements in AI",
num_results=5
)
# With custom parameters
results = await tool.execute(
query="Python async programming best practices",
num_results=10
)
Search Results for: "query" (Time: 0.45s, Approx. Total: 1,000,000)
Result 1:
Title: Result Title
Link: https://example.com/article
Snippet: Article snippet text...
Result 2:
...
Error: Web search API request failed with status 403. Detail: API key invalid
Mock Search Results for: "query" (API Key/ID Not Configured or httpx missing)
Result 1:
Title: Mock Result 1 - query
Link: https://example.com/mocksearch?q=query&result=1
Snippet: This is mock search result...
When API keys are not configured:
Endpoint: https://www.googleapis.com/customsearch/v1
Parameters:
key: API keycx: Search Engine IDq: Search querynum: Number of results (1-10)Rate Limits:
# In agent plan
plan = [
{
"action": "search_web",
"query": "latest Python async features",
"num_results": 5
}
]
The WebSearchTool can be used by:
results = await tool.execute("mindX autonomous agents")
print(results)
results = await tool.execute(
query="autonomous AI systems 2024",
num_results=10
)
httpx: Async HTTP client librarycore.bdi_agent.BaseTool: Base tool classutils.config.Config: Configuration managementutils.logging_config.get_logger: Logging utilitytry:
response = await self.http_client.get(search_url, params=params)
response.raise_for_status()
results_json = response.json()
return self._format_google_search_results(results_json, query)
except httpx.TimeoutException as e:
return f"Error: Web search request timed out. ({e})"
except httpx.HTTPStatusError as e:
return f"Error: Web search API request failed with status {e.response.status_code}"
except Exception as e:
return f"Error: Unexpected error during web search - {type(e).__name__}: {e}"
The tool properly closes HTTP client on shutdown:
async def shutdown(self):
if self.http_client and not self.http_client.is_closed:
await self.http_client.aclose()