Pi coding agents need reliable web search to look up documentation and verify solutions. A single search provider going down breaks the agent. This tutorial configures Pi Agent with Scavio as the primary search backend and adds fallback providers for reliability. The agent automatically switches providers on failure.
Prerequisites
- Python 3.8+
- requests library
- A Scavio API key from scavio.dev
- Pi Agent or similar coding agent
Walkthrough
Step 1: Build the multi-backend search client
Create a search client that routes queries through multiple providers in priority order.
import os, requests, time, json
class MultiSearchClient:
def __init__(self):
self.providers = [
{
'name': 'scavio',
'fn': self._search_scavio,
'failures': 0,
'last_fail': 0,
},
]
self.scavio_key = os.environ.get('SCAVIO_API_KEY', '')
def _search_scavio(self, query):
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': self.scavio_key, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us'}, timeout=10)
resp.raise_for_status()
data = resp.json()
return [{'title': r.get('title', ''), 'url': r.get('link', ''),
'snippet': r.get('snippet', '')} for r in data.get('organic_results', [])]
def search(self, query, num_results=5):
for provider in self.providers:
if provider['failures'] >= 3 and time.time() - provider['last_fail'] < 60:
print(f' Skipping {provider["name"]} (cooldown)')
continue
try:
results = provider['fn'](query)[:num_results]
provider['failures'] = 0
return {'provider': provider['name'], 'results': results, 'count': len(results)}
except Exception as e:
provider['failures'] += 1
provider['last_fail'] = time.time()
print(f' {provider["name"]} failed: {str(e)[:40]}')
return {'provider': 'none', 'results': [], 'count': 0}
client = MultiSearchClient()
result = client.search('python FastAPI async tutorial')
print(f'Provider: {result["provider"]} | Results: {result["count"]}')
for r in result['results'][:3]:
print(f' {r["title"][:50]}')Step 2: Integrate with Pi Agent tool system
Register the multi-search client as a tool the Pi Agent can call.
client = MultiSearchClient()
PI_SEARCH_TOOL = {
'name': 'web_search',
'description': 'Search the web with automatic provider failover. Use for docs, APIs, errors, or any question needing current data.',
'parameters': {
'type': 'object',
'properties': {
'query': {'type': 'string', 'description': 'Search query'},
'num_results': {'type': 'integer', 'default': 5}
},
'required': ['query']
}
}
def pi_search(query, num_results=5):
"""Pi Agent search tool with multi-provider failover."""
result = client.search(query, num_results)
# Format for agent consumption
formatted = f'Search results for "{query}" (via {result["provider"]}):\n'
for i, r in enumerate(result['results'], 1):
formatted += f'{i}. {r["title"]}\n URL: {r["url"]}\n {r["snippet"]}\n\n'
return formatted
# Test agent-formatted output
print(pi_search('how to handle database migrations in FastAPI'))Step 3: Add platform-specific search for coding tasks
Extend the tool to search specific platforms like Reddit and YouTube for coding help.
def pi_search_platform(query, platform=None, num_results=5):
"""Search specific platforms for more targeted results."""
body = {'query': query, 'country_code': 'us'}
if platform:
body['platform'] = platform
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': client.scavio_key, 'Content-Type': 'application/json'},
json=body, timeout=10)
resp.raise_for_status()
data = resp.json()
return [{'title': r.get('title', ''), 'url': r.get('link', ''),
'snippet': r.get('snippet', '')} for r in data.get('organic_results', [])[:num_results]]
except Exception:
return []
# Pi Agent can now search specific platforms
print('=== Google (docs) ===')
for r in pi_search_platform('FastAPI async SQLAlchemy')[:2]:
print(f' {r["title"][:50]}')
print('\n=== Reddit (community answers) ===')
for r in pi_search_platform('FastAPI async SQLAlchemy', platform='reddit')[:2]:
print(f' {r["title"][:50]}')
print('\n=== YouTube (video tutorials) ===')
for r in pi_search_platform('FastAPI async SQLAlchemy', platform='youtube')[:2]:
print(f' {r["title"][:50]}')
print(f'\nTotal cost: $0.015 (3 platform searches)')
print(f'Pi Agent now has: web, reddit, youtube search')Python Example
import os, requests
SH = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
def pi_search(query):
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': query, 'country_code': 'us'}, timeout=10).json()
return [{'title': r['title'], 'url': r['link']} for r in data.get('organic_results', [])[:5]]
for r in pi_search('FastAPI tutorial 2026'):
print(f'{r["title"][:50]}')JavaScript Example
const SH = { 'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json' };
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: SH,
body: JSON.stringify({ query: 'FastAPI tutorial 2026', country_code: 'us' })
}).then(r => r.json());
(data.organic_results || []).slice(0, 5).forEach(r => console.log(r.title));Expected Output
Provider: scavio | Results: 5
FastAPI with Async SQLAlchemy - Complete Guide
Python Async Database Tutorial 2026
Search results for "how to handle database migrations in FastAPI" (via scavio):
1. Alembic Migrations with FastAPI - Official Docs
URL: https://fastapi.tiangolo.com/...
Learn how to set up Alembic for database migrations...
=== Google (docs) ===
FastAPI Async SQLAlchemy - Official Documentation
=== Reddit (community answers) ===
r/FastAPI - Best practices for async SQLAlchemy
=== YouTube (video tutorials) ===
FastAPI + Async SQLAlchemy Full Tutorial 2026
Total cost: $0.015 (3 platform searches)