Unify Multiple Search APIs with Abstract Class, Return Errors Instead of Raising
· 4 min read
Encountered this issue while building an AI Agent platform for a client: needed to support multiple search providers (Tavily, Serper, Brave, Bing) while ensuring tool call failures don't interrupt the Agent's conversation flow.
TL;DR
- Define
SearchProviderabstract base class +SearchResultdata model for unified interface and output - Each provider inherits the base class, implements
search()method with field mapping - Key design: Return
SearchResultwith error info on failure, never raise exceptions
The Problem
Direct calls to different search APIs look like this:
# Tavily: POST request, results[].url
response = await client.post("https://api.tavily.com/search", ...)
# Serper: POST request, organic[].link
response = await client.post("https://google.serper.dev/search", ...)
# Brave: GET request, web.results[].description
response = await client.get("https://api.search.brave.com/res/v1/web/search", ...)
# Bing: GET request, webPages.value[].snippet
response = await client.get("https://api.bing.microsoft.com/v7.0/search", ...)
Issues:
- Request methods, auth headers, and response structures vary
- Switching providers requires changing caller code
raise Exceptioninterrupts AI Agent's streaming conversation
Root Cause
- Missing abstraction layer: Caller directly depends on concrete implementations, violating dependency inversion
- Inconsistent error handling: Exceptions propagate up the call stack, crashing the entire streaming flow
For AI Agent tool calls, the Agent needs to decide whether to retry, use another tool, or explain to the user—not just crash.
Solution
1. Define Abstract Base Class and Data Model
# base.py
from abc import ABC, abstractmethod
from typing import List
from pydantic import BaseModel
class SearchResult(BaseModel):
"""Unified search result."""
title: str
link: str
snippet: str
class SearchProvider(ABC):
"""Base class for search providers."""
def __init__(self, api_key: str):
self.api_key = api_key
@abstractmethod
async def search(self, query: str, max_results: int = 5) -> List[SearchResult]:
"""Execute search and return results."""
pass
2. Implement Concrete Providers
Tavily (AI-optimized search, supports rate limit / quota error codes):
# tavily.py
import httpx
import logging
from typing import List
from .base import SearchProvider, SearchResult
logger = logging.getLogger(__name__)
class TavilySearch(SearchProvider):
"""Tavily Search API implementation."""
async def search(self, query: str, max_results: int = 5) -> List[SearchResult]:
try:
async with httpx.AsyncClient(timeout=15.0) as client:
response = await client.post(
"https://api.tavily.com/search",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"query": query,
"max_results": max_results,
"search_depth": "basic"
}
)
# Return SearchResult on error, never raise
if response.status_code == 429:
return [SearchResult(
title="Rate Limited",
link="",
snippet="Search quota exceeded. Please try again later."
)]
if response.status_code == 401:
return [SearchResult(
title="Auth Error",
link="",
snippet="Search API key is invalid."
)]
if response.status_code == 402:
return [SearchResult(
title="Quota Exceeded",
link="",
snippet="Monthly search quota depleted."
)]
response.raise_for_status()
data = response.json()
# Field mapping: Tavily's url -> unified link
results = []
for item in data.get("results", [])[:max_results]:
results.append(SearchResult(
title=item.get("title", ""),
link=item.get("url", ""),
snippet=item.get("content", "")
))
return results
except httpx.TimeoutException:
logger.warning(f"Tavily API timeout: {query[:50]}")
return [SearchResult(title="Timeout", link="", snippet="Search timed out.")]
except Exception as e:
logger.error(f"Tavily search error: {e}")
return [SearchResult(title="Error", link="", snippet=f"Search failed: {str(e)}")]
Serper (Google Search API):
# serper.py
class SerperSearch(SearchProvider):
"""Serper (Google Search) API implementation."""
async def search(self, query: str, max_results: int = 5) -> List[SearchResult]:
try:
async with httpx.AsyncClient(timeout=15.0) as client:
response = await client.post(
"https://google.serper.dev/search",
headers={"X-API-KEY": self.api_key, "Content-Type": "application/json"},
json={"q": query, "num": max_results}
)
if response.status_code == 401:
return [SearchResult(title="Auth Error", link="", snippet="Serper API key is invalid.")]
response.raise_for_status()
data = response.json()
# Field mapping: Serper's organic[].link -> unified link
results = []
for item in data.get("organic", [])[:max_results]:
results.append(SearchResult(
title=item.get("title", ""),
link=item.get("link", ""),
snippet=item.get("snippet", "")
))
return results
except httpx.TimeoutException:
return [SearchResult(title="Timeout", link="", snippet="Search timed out.")]
except Exception as e:
return [SearchResult(title="Error", link="", snippet=f"Search failed: {str(e)}")]
Brave and Bing implementations are similar, differing in request method and response field mapping.
3. Caller Usage
# Depend on abstraction only
async def execute_search(provider: SearchProvider, query: str) -> List[SearchResult]:
results = await provider.search(query)
# Check for errors (via title or snippet)
if results and not results[0].link:
error_msg = results[0].snippet
# Agent can decide next action based on error info
return f"Search failed: {error_msg}"
return results
# Switch providers by changing instance only
provider = TavilySearch(api_key="xxx")
# provider = SerperSearch(api_key="xxx")
results = await execute_search(provider, "Python async best practices")
Key Design Decisions
| Decision | Reason |
|---|---|
Return SearchResult on error instead of raise | AI Agent conversations are streaming flows; exceptions interrupt everything |
Use Pydantic BaseModel for output | Auto-validation + IDE hints + JSON serialization |
Use ABC instead of Protocol | Need shared __init__ logic (api_key storage) |
| Unified 15-second timeout | Search is UX-critical; can't be too slow |
Interested in similar solutions? Get in touch