La mayoría de los agentes de IA no ven los datos en tiempo real. Alucinan precios, inventan nombres de productos y se pierden los acontecimientos actuales. Agregar una herramienta de búsqueda que cubra seis plataformas (Google, Amazon, YouTube, Walmart, Reddit, TikTok) a través de un punto final API soluciona este problema. Scavio proporciona las seis plataformas a través de una única solicitud POST a $0,005 por crédito. Este tutorial muestra cómo conectar la búsqueda multiplataforma a cualquier marco de agente, desde llamadas a funciones sin formato hasta LangChain y CrewAI.
Requisitos previos
- Python 3.9+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Una clave API de LLM (OpenAI, Anthropic u Ollama local)
Guia paso a paso
Paso 1: Cree la herramienta de búsqueda unificada
Cree una función única que busque cualquiera de las seis plataformas según la consulta. El agente decide a qué plataforma dirigirse mediante prefijos de consulta.
Python
import os, requests
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
SEARCH_URL = 'https://api.scavio.dev/api/v1/search'
TT_URL = 'https://api.scavio.dev/api/v1/tiktok'
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
TT_H = {'Authorization': f'Bearer {SCAVIO_KEY}', 'Content-Type': 'application/json'}
def search_platform(query: str, platform: str = 'google', num: int = 5) -> list:
"""Search any platform. Returns normalized results."""
if platform == 'tiktok':
resp = requests.post(f'{TT_URL}/search/videos', headers=TT_H,
json={'keyword': query, 'count': num, 'cursor': 0})
videos = resp.json().get('data', {}).get('videos', [])
return [{'title': v.get('desc', '')[:100], 'url': f'tiktok.com/@{v.get("author",{}).get("uniqueId","")}',
'snippet': f'Plays: {v.get("stats",{}).get("playCount",0):,}', 'platform': 'tiktok'} for v in videos]
# All other platforms use the search endpoint with site: prefix
site_map = {'amazon': 'amazon.com', 'youtube': 'youtube.com',
'walmart': 'walmart.com', 'reddit': 'reddit.com'}
q = f'site:{site_map[platform]} {query}' if platform in site_map else query
resp = requests.post(SEARCH_URL, headers=H,
json={'query': q, 'country_code': 'us', 'num_results': num})
return [{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', ''),
'platform': platform} for r in resp.json().get('organic_results', [])]
for p in ['google', 'amazon', 'reddit']:
results = search_platform('wireless earbuds', platform=p, num=2)
print(f'{p}: {len(results)} results')
for r in results:
print(f' {r["title"][:60]}')Paso 2: Crear la definición de la herramienta agente
Defina la herramienta de búsqueda con un esquema que cualquier marco de agente pueda utilizar. La herramienta acepta una consulta y un parámetro de plataforma opcional.
Python
TOOL_DEFINITION = {
'type': 'function',
'function': {
'name': 'search',
'description': ('Search across Google, Amazon, YouTube, Walmart, Reddit, or TikTok. '
'Use platform parameter to target a specific source.'),
'parameters': {
'type': 'object',
'properties': {
'query': {'type': 'string', 'description': 'The search query'},
'platform': {
'type': 'string',
'enum': ['google', 'amazon', 'youtube', 'walmart', 'reddit', 'tiktok'],
'description': 'Platform to search. Default: google'
},
'num_results': {'type': 'integer', 'description': 'Number of results (1-10)', 'default': 5}
},
'required': ['query']
}
}
}
def handle_tool_call(name: str, args: dict) -> str:
"""Execute agent tool calls."""
if name == 'search':
results = search_platform(
query=args['query'],
platform=args.get('platform', 'google'),
num=args.get('num_results', 5)
)
return '\n'.join(f'[{i+1}] {r["title"]} ({r["platform"]})\n {r["snippet"]}'
for i, r in enumerate(results))
return 'Unknown tool'
print(handle_tool_call('search', {'query': 'best earbuds', 'platform': 'reddit'}))Paso 3: Conéctese a un bucle de agente compatible con OpenAI
Cree el bucle del agente que envía definiciones de herramientas al LLM, ejecuta llamadas a herramientas y envía resultados. Funciona con OpenAI, Anthropic mediante adaptador o modelos locales.
Python
import json
LLM_URL = os.environ.get('LLM_URL', 'http://localhost:11434/v1/chat/completions')
LLM_KEY = os.environ.get('LLM_API_KEY', 'ollama')
def agent_loop(user_query: str, max_turns: int = 5) -> str:
messages = [
{'role': 'system', 'content': (
'You are a research assistant with access to 6 search platforms: '
'Google, Amazon, YouTube, Walmart, Reddit, and TikTok. '
'Use the search tool to find real-time data. Search multiple platforms '
'when the user needs a comprehensive answer.'
)},
{'role': 'user', 'content': user_query}
]
for turn in range(max_turns):
resp = requests.post(LLM_URL, headers={
'Authorization': f'Bearer {LLM_KEY}', 'Content-Type': 'application/json'
}, json={'model': 'llama3', 'messages': messages,
'tools': [TOOL_DEFINITION], 'max_tokens': 1024})
choice = resp.json()['choices'][0]
msg = choice['message']
messages.append(msg)
if not msg.get('tool_calls'):
return msg['content']
for tc in msg['tool_calls']:
args = json.loads(tc['function']['arguments']) if isinstance(tc['function']['arguments'], str) else tc['function']['arguments']
result = handle_tool_call(tc['function']['name'], args)
messages.append({'role': 'tool', 'tool_call_id': tc['id'], 'content': result})
return messages[-1].get('content', 'Max turns reached')
answer = agent_loop('Compare wireless earbuds prices on Amazon vs Walmart')
print(answer)Paso 4: Añadir búsqueda paralela multiplataforma
Para consultas completas, busque en varias plataformas en paralelo mediante subprocesos. Esto reduce la latencia cuando el agente necesita datos de varias fuentes a la vez.
Python
from concurrent.futures import ThreadPoolExecutor, as_completed
def multi_platform_search(query: str, platforms: list[str] = None) -> dict:
"""Search multiple platforms in parallel."""
if platforms is None:
platforms = ['google', 'amazon', 'reddit']
results = {}
with ThreadPoolExecutor(max_workers=len(platforms)) as executor:
futures = {executor.submit(search_platform, query, p, 3): p
for p in platforms}
for future in as_completed(futures):
platform = futures[future]
try:
results[platform] = future.result()
except Exception as e:
results[platform] = [{'title': f'Error: {e}', 'url': '', 'snippet': '', 'platform': platform}]
return results
def format_multi_results(results: dict) -> str:
lines = []
idx = 1
for platform, items in results.items():
lines.append(f'\n--- {platform.upper()} ---')
for r in items:
lines.append(f'[{idx}] {r["title"]}')
if r['snippet']:
lines.append(f' {r["snippet"][:150]}')
idx += 1
return '\n'.join(lines)
results = multi_platform_search('noise cancelling headphones')
print(format_multi_results(results))
print(f'\nTotal cost: ${sum(len(v) for v in results.values()) * 0.005:.3f}')Ejemplo en Python
Python
import os, requests, json
from concurrent.futures import ThreadPoolExecutor
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
URL = 'https://api.scavio.dev/api/v1/search'
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
def search(query, platform='google', num=3):
site_map = {'amazon': 'amazon.com', 'youtube': 'youtube.com',
'walmart': 'walmart.com', 'reddit': 'reddit.com'}
q = f'site:{site_map[platform]} {query}' if platform in site_map else query
resp = requests.post(URL, headers=H, json={'query': q, 'country_code': 'us', 'num_results': num})
return [{'title': r['title'], 'url': r['link'], 'platform': platform}
for r in resp.json().get('organic_results', [])]
def multi_search(query, platforms=['google', 'amazon', 'reddit']):
with ThreadPoolExecutor(max_workers=3) as ex:
futures = {ex.submit(search, query, p): p for p in platforms}
return {futures[f]: f.result() for f in futures}
results = multi_search('wireless earbuds 2026')
for p, items in results.items():
print(f'{p}: {len(items)} results')
for r in items:
print(f' {r["title"][:60]}')Ejemplo en JavaScript
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
const URL = 'https://api.scavio.dev/api/v1/search';
async function search(query, platform = 'google', num = 3) {
const siteMap = { amazon: 'amazon.com', youtube: 'youtube.com', walmart: 'walmart.com', reddit: 'reddit.com' };
const q = siteMap[platform] ? `site:${siteMap[platform]} ${query}` : query;
const resp = await fetch(URL, {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query: q, country_code: 'us', num_results: num })
});
return ((await resp.json()).organic_results || []).map(r => ({ title: r.title, url: r.link, platform }));
}
async function multiSearch(query, platforms = ['google', 'amazon', 'reddit']) {
const results = await Promise.all(platforms.map(p => search(query, p)));
return Object.fromEntries(platforms.map((p, i) => [p, results[i]]));
}
multiSearch('wireless earbuds 2026').then(r => {
for (const [p, items] of Object.entries(r)) {
console.log(`${p}: ${items.length} results`);
items.forEach(i => console.log(` ${i.title.slice(0, 60)}`));
}
});Salida esperada
JSON
google: 3 results
Best Wireless Earbuds 2026: Top Picks Reviewed
Wireless Earbuds Buying Guide - What to Look For
Top 10 Earbuds Under $100 in 2026
amazon: 3 results
Sony WF-1000XM6 Wireless Noise Cancelling Earbuds
Apple AirPods Pro 3 with Adaptive Audio
Samsung Galaxy Buds4 Pro
reddit: 3 results
r/headphones - Best wireless earbuds mid-2026?
r/BudgetAudiophile - Earbuds under $50 megathread
r/apple - AirPods Pro 3 vs Sony XM6 honest comparison