Improve web search and error handling

- Add DuckDuckGo search fallback when Brave API key is not available
  - Web search now works without requiring an API key
  - Falls back to DuckDuckGo if BRAVE_API_KEY is not set
  - Maintains backward compatibility with Brave API when key is provided

- Improve error handling in agent CLI command
  - Better exception handling with traceback display
  - Prevents crashes from showing incomplete error messages
  - Improves debugging experience
This commit is contained in:
Tanya 2026-02-18 12:41:11 -05:00
parent 7961bf1360
commit 9c858699f3
2 changed files with 94 additions and 7 deletions

View File

@ -44,7 +44,7 @@ def _validate_url(url: str) -> tuple[bool, str]:
class WebSearchTool(Tool):
"""Search the web using Brave Search API."""
"""Search the web using DuckDuckGo (free, no API key required)."""
name = "web_search"
description = "Search the web. Returns titles, URLs, and snippets."
@ -58,13 +58,20 @@ class WebSearchTool(Tool):
}
def __init__(self, api_key: str | None = None, max_results: int = 5):
# Keep api_key parameter for backward compatibility, but use DuckDuckGo if not provided
self.api_key = api_key or os.environ.get("BRAVE_API_KEY", "")
self.max_results = max_results
self.use_brave = bool(self.api_key)
async def execute(self, query: str, count: int | None = None, **kwargs: Any) -> str:
if not self.api_key:
return "Error: BRAVE_API_KEY not configured"
# Try Brave API if key is available, otherwise use DuckDuckGo
if self.use_brave:
return await self._brave_search(query, count)
else:
return await self._duckduckgo_search(query, count)
async def _brave_search(self, query: str, count: int | None = None) -> str:
"""Search using Brave API (requires API key)."""
try:
n = min(max(count or self.max_results, 1), 10)
async with httpx.AsyncClient() as client:
@ -88,6 +95,79 @@ class WebSearchTool(Tool):
return "\n".join(lines)
except Exception as e:
return f"Error: {e}"
async def _duckduckgo_search(self, query: str, count: int | None = None) -> str:
"""Search using DuckDuckGo (free, no API key)."""
try:
n = min(max(count or self.max_results, 1), 10)
# Try using duckduckgo_search library if available
try:
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = []
for r in ddgs.text(query, max_results=n):
results.append({
"title": r.get("title", ""),
"url": r.get("href", ""),
"description": r.get("body", "")
})
if not results:
return f"No results found for: {query}"
lines = [f"Results for: {query}\n"]
for i, item in enumerate(results, 1):
lines.append(f"{i}. {item['title']}\n {item['url']}")
if item['description']:
lines.append(f" {item['description']}")
return "\n".join(lines)
except ImportError:
# Fallback: use DuckDuckGo instant answer API (simpler, but limited)
async with httpx.AsyncClient(
follow_redirects=True,
timeout=15.0
) as client:
# Use DuckDuckGo instant answer API (no key needed)
url = "https://api.duckduckgo.com/"
r = await client.get(
url,
params={"q": query, "format": "json", "no_html": "1", "skip_disambig": "1"},
headers={"User-Agent": USER_AGENT},
)
r.raise_for_status()
data = r.json()
results = []
# Get RelatedTopics (search results)
if "RelatedTopics" in data:
for topic in data["RelatedTopics"][:n]:
if "Text" in topic and "FirstURL" in topic:
results.append({
"title": topic.get("Text", "").split(" - ")[0] if " - " in topic.get("Text", "") else topic.get("Text", "")[:50],
"url": topic.get("FirstURL", ""),
"description": topic.get("Text", "")
})
# Also check AbstractText for direct answer
if "AbstractText" in data and data["AbstractText"]:
results.insert(0, {
"title": data.get("Heading", query),
"url": data.get("AbstractURL", ""),
"description": data.get("AbstractText", "")
})
if not results:
return f"No results found for: {query}. Try installing 'duckduckgo-search' package for better results: pip install duckduckgo-search"
lines = [f"Results for: {query}\n"]
for i, item in enumerate(results[:n], 1):
lines.append(f"{i}. {item['title']}\n {item['url']}")
if item['description']:
lines.append(f" {item['description']}")
return "\n".join(lines)
except Exception as e:
return f"Error searching: {e}. Try installing 'duckduckgo-search' package: pip install duckduckgo-search"
class WebFetchTool(Tool):

View File

@ -494,9 +494,16 @@ def agent(
if message:
# Single message mode
async def run_once():
with _thinking_ctx():
response = await agent_loop.process_direct(message, session_id)
_print_agent_response(response, render_markdown=markdown)
try:
with _thinking_ctx():
response = await agent_loop.process_direct(message, session_id)
# response is a string (content) from process_direct
_print_agent_response(response or "", render_markdown=markdown)
except Exception as e:
import traceback
console.print(f"[red]Error: {e}[/red]")
console.print(f"[dim]{traceback.format_exc()}[/dim]")
raise
asyncio.run(run_once())
else: