Los agentes de codificación como Aider, Continuar y Cursor necesitan búsquedas web para búsquedas de documentación, validación de paquetes y verificaciones de referencias de API. Un único backend de búsqueda crea un único punto de error. Este tutorial crea una extensión de búsqueda multibackend que dispara solicitudes paralelas a Scavio, Brave y Tavily, devuelve la respuesta más rápida y recurre al siguiente proveedor si uno falla. La extensión expone una interfaz unificada que cualquier agente de codificación puede consumir.
Requisitos previos
- Python 3.9+ instalado
- solicita biblioteca instalada
- Claves API para Scavio y al menos un proveedor de respaldo
- Un agente de codificación que admite extensiones de herramientas personalizadas
Guia paso a paso
Paso 1: Configurar el ejecutor de búsqueda paralela
Utilice concurrent.futures para enviar solicitudes de búsqueda a varios proveedores simultáneamente. La primera respuesta exitosa gana.
import os, requests, time
from concurrent.futures import ThreadPoolExecutor, as_completed
SCAVIO_KEY = os.environ.get('SCAVIO_API_KEY', '')
BRAVE_KEY = os.environ.get('BRAVE_API_KEY', '')
TAVILY_KEY = os.environ.get('TAVILY_API_KEY', '')
def search_scavio(query: str) -> list:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': 5}, timeout=8)
resp.raise_for_status()
return [{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', ''), 'source': 'scavio'}
for r in resp.json().get('organic_results', [])]
def search_brave(query: str) -> list:
resp = requests.get('https://api.search.brave.com/res/v1/web/search',
headers={'X-Subscription-Token': BRAVE_KEY},
params={'q': query, 'count': 5}, timeout=8)
resp.raise_for_status()
return [{'title': r['title'], 'url': r['url'], 'snippet': r.get('description', ''), 'source': 'brave'}
for r in resp.json().get('web', {}).get('results', [])]Paso 2: Construya el ejecutor paralelo con conmutación por error
Ejecute todos los proveedores en paralelo y devuelva el primer resultado exitoso. Registre qué proveedor atendió la solicitud de depuración.
PROVIDERS = [
('scavio', search_scavio),
('brave', search_brave),
]
def parallel_search(query: str, timeout: float = 10.0) -> dict:
"""Search multiple providers in parallel, return first success."""
with ThreadPoolExecutor(max_workers=len(PROVIDERS)) as executor:
futures = {executor.submit(fn, query): name for name, fn in PROVIDERS}
for future in as_completed(futures, timeout=timeout):
provider = futures[future]
try:
results = future.result()
if results:
return {'results': results, 'provider': provider, 'query': query}
except Exception as e:
print(f'{provider} failed: {e}')
return {'results': [], 'provider': 'none', 'query': query}
result = parallel_search('python requests library documentation')
print(f'Provider: {result["provider"]}')
for r in result['results'][:3]:
print(f' {r["title"]}')
print(f' {r["url"]}')Paso 3: Dar formato a los resultados para los agentes codificadores
Formatee los resultados de búsqueda de una manera que los agentes de codificación puedan analizar: incluya URL como referencia, fragmentos centrados en el código e indicadores de registro de paquetes.
def format_for_agent(search_result: dict) -> str:
lines = [f'Search results for: {search_result["query"]}', '']
for i, r in enumerate(search_result['results'], 1):
is_docs = any(d in r['url'] for d in ['docs.', 'readthedocs', 'pypi.org', 'npmjs.com', 'github.com'])
prefix = '[DOCS] ' if is_docs else ''
lines.append(f'{i}. {prefix}{r["title"]}')
lines.append(f' URL: {r["url"]}')
if r['snippet']:
lines.append(f' {r["snippet"][:150]}')
lines.append('')
lines.append(f'(served by {search_result["provider"]})')
return '\n'.join(lines)
formatted = format_for_agent(result)
print(formatted)Paso 4: Exponer como extensión de herramienta
Cree una definición de herramienta que cualquier agente de codificación compatible con MCP pueda utilizar. La herramienta acepta una consulta y devuelve resultados de búsqueda formateados.
TOOL_DEFINITION = {
'name': 'multi_search',
'description': 'Search the web using multiple providers with automatic failover. Useful for documentation, package lookups, and API references.',
'inputSchema': {
'type': 'object',
'properties': {
'query': {'type': 'string', 'description': 'Search query'},
'focus': {'type': 'string', 'enum': ['docs', 'code', 'general'], 'description': 'Search focus area'}
},
'required': ['query']
}
}
def handle_tool_call(args: dict) -> str:
query = args['query']
focus = args.get('focus', 'general')
if focus == 'docs':
query = f'{query} documentation'
elif focus == 'code':
query = f'{query} code example'
result = parallel_search(query)
return format_for_agent(result)
# Test
print(handle_tool_call({'query': 'httpx async client', 'focus': 'docs'}))Ejemplo en Python
import os, requests
from concurrent.futures import ThreadPoolExecutor, as_completed
SCAVIO_KEY = os.environ.get('SCAVIO_API_KEY', '')
BRAVE_KEY = os.environ.get('BRAVE_API_KEY', '')
def search_scavio(q):
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': q, 'country_code': 'us', 'num_results': 5}, timeout=8)
return [{'title': x['title'], 'url': x['link']} for x in r.json().get('organic_results', [])]
def search_brave(q):
r = requests.get('https://api.search.brave.com/res/v1/web/search',
headers={'X-Subscription-Token': BRAVE_KEY}, params={'q': q, 'count': 5}, timeout=8)
return [{'title': x['title'], 'url': x['url']} for x in r.json().get('web', {}).get('results', [])]
def multi_search(query):
with ThreadPoolExecutor(max_workers=2) as ex:
futs = {ex.submit(fn, query): n for n, fn in [('scavio', search_scavio), ('brave', search_brave)]}
for f in as_completed(futs, timeout=10):
try:
results = f.result()
if results:
print(f'Served by {futs[f]}')
return results
except: pass
return []
for r in multi_search('httpx documentation')[:3]:
print(f' {r["title"]}')Ejemplo en JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
const BRAVE_KEY = process.env.BRAVE_API_KEY;
async function searchScavio(query) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us', num_results: 5 })
});
return (await resp.json()).organic_results?.map(r => ({ title: r.title, url: r.link })) || [];
}
async function searchBrave(query) {
const resp = await fetch(`https://api.search.brave.com/res/v1/web/search?q=${encodeURIComponent(query)}&count=5`, {
headers: { 'X-Subscription-Token': BRAVE_KEY }
});
return (await resp.json()).web?.results?.map(r => ({ title: r.title, url: r.url })) || [];
}
async function multiSearch(query) {
const result = await Promise.any([searchScavio(query), searchBrave(query)]);
result.slice(0, 3).forEach(r => console.log(` ${r.title}`));
}
multiSearch('httpx async documentation');Salida esperada
Provider: scavio
Python Requests Library Documentation
https://docs.python-requests.org/en/latest/
httpx - A Modern HTTP Client for Python
https://www.python-httpx.org/
1. [DOCS] Python Requests Library Documentation
URL: https://docs.python-requests.org/en/latest/
Requests is an elegant and simple HTTP library for Python...
(served by scavio)