La investigación de mercados manual consume horas cada semana: revisar las páginas de la competencia, rastrear la clasificación de las palabras clave, escanear Reddit en busca de menciones de marcas. Este tutorial crea un agente de investigación automatizado que maneja los tres utilizando la API de búsqueda de Scavio, almacena los hallazgos en informes estructurados y cuesta menos de $1 al mes para el monitoreo diario de 10 competidores y 50 palabras clave.
Requisitos previos
- Python 3.11+
- Una clave API de Scavio de https://scavio.dev
- SQLite3 (incluido con Python)
- Opcional: un webhook de Slack o Discord para alertas
Guia paso a paso
Paso 1: Establecer la configuración del agente de investigación
Defina sus competidores, las palabras clave rastreadas y los objetivos de seguimiento en un archivo de configuración. El agente lee esto al inicio para saber qué investigar.
import json
from pathlib import Path
from dataclasses import dataclass, field
from datetime import date
@dataclass
class ResearchConfig:
brand: str
competitors: list[str]
keywords: list[str]
reddit_queries: list[str]
domain: str
@classmethod
def from_file(cls, path: str) -> "ResearchConfig":
data = json.loads(Path(path).read_text())
return cls(**data)
# Example config
config = ResearchConfig(
brand="Scavio",
domain="scavio.dev",
competitors=[
"serpapi.com",
"serper.dev",
"brightdata.com",
],
keywords=[
"best search API for agents 2026",
"cheap web scraping API",
"MCP search server",
],
reddit_queries=[
"search API recommendation",
"web scraping API affordable",
"MCP tools search",
]
)Paso 2: Monitorear a los competidores y detectar cambios en la página
Busque las páginas clave de cada competidor y compárelas con instantáneas anteriores. Marque páginas nuevas, páginas eliminadas y cambios de contenido importantes.
import httpx
import sqlite3
SCAVIO_API_KEY = "your-api-key"
DB_PATH = Path("research.db")
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.executescript("""
CREATE TABLE IF NOT EXISTS competitor_pages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
competitor TEXT, url TEXT, title TEXT,
snippet TEXT, first_seen TEXT, last_seen TEXT
);
CREATE TABLE IF NOT EXISTS keyword_ranks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
keyword TEXT, domain TEXT, position INTEGER,
url TEXT, checked_at TEXT
);
CREATE TABLE IF NOT EXISTS reddit_mentions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query TEXT, url TEXT, title TEXT,
snippet TEXT, found_at TEXT
);
""")
conn.commit()
conn.close()
async def monitor_competitor(client: httpx.AsyncClient, competitor: str) -> list[dict]:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": f"site:{competitor}", "num_results": 15}
)
resp.raise_for_status()
results = resp.json().get("results", [])
conn = sqlite3.connect(DB_PATH)
today = date.today().isoformat()
new_pages = []
for r in results:
url = r.get("url", "")
existing = conn.execute(
"SELECT id FROM competitor_pages WHERE url = ? AND competitor = ?",
(url, competitor)
).fetchone()
if existing:
conn.execute("UPDATE competitor_pages SET last_seen = ? WHERE id = ?", (today, existing[0]))
else:
conn.execute(
"INSERT INTO competitor_pages VALUES (NULL,?,?,?,?,?,?)",
(competitor, url, r.get("title", ""), r.get("description", "")[:300], today, today)
)
new_pages.append({"url": url, "title": r.get("title", "")})
conn.commit()
conn.close()
return new_pagesPaso 3: Realice un seguimiento de las clasificaciones de palabras clave y escanee Reddit
Verifique las posiciones de las palabras clave y busque en Reddit menciones de marca. Ambos utilizan la misma API de búsqueda Scavio con diferentes patrones de consulta.
async def track_keywords(client: httpx.AsyncClient, keywords: list[str], domain: str):
conn = sqlite3.connect(DB_PATH)
today = date.today().isoformat()
for kw in keywords:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": kw, "num_results": 10}
)
resp.raise_for_status()
position = None
url = None
for i, r in enumerate(resp.json().get("results", [])):
if domain in r.get("url", ""):
position = i + 1
url = r["url"]
break
conn.execute(
"INSERT INTO keyword_ranks VALUES (NULL,?,?,?,?,?)",
(kw, domain, position, url, today)
)
conn.commit()
conn.close()
async def scan_reddit(client: httpx.AsyncClient, queries: list[str]) -> list[dict]:
conn = sqlite3.connect(DB_PATH)
today = date.today().isoformat()
new_mentions = []
for q in queries:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": f"site:reddit.com {q}", "num_results": 10}
)
resp.raise_for_status()
for r in resp.json().get("results", []):
url = r.get("url", "")
existing = conn.execute(
"SELECT id FROM reddit_mentions WHERE url = ?", (url,)
).fetchone()
if not existing:
conn.execute(
"INSERT INTO reddit_mentions VALUES (NULL,?,?,?,?,?)",
(q, url, r.get("title", ""), r.get("description", "")[:300], today)
)
new_mentions.append({"url": url, "title": r.get("title", ""), "query": q})
conn.commit()
conn.close()
return new_mentionsPaso 4: Generar el informe diario de investigación
Ejecute las tres tareas de investigación y compile un informe diario con nuevas páginas de la competencia, cambios en la clasificación y menciones en Reddit.
import asyncio
async def daily_research(config: ResearchConfig) -> dict:
init_db()
report = {
"date": date.today().isoformat(),
"new_competitor_pages": [],
"keyword_rankings": [],
"reddit_mentions": [],
"credits_used": 0
}
async with httpx.AsyncClient(timeout=15) as client:
# Monitor competitors
for comp in config.competitors:
new_pages = await monitor_competitor(client, comp)
report["new_competitor_pages"].extend(
[{"competitor": comp, **p} for p in new_pages]
)
report["credits_used"] += 1
# Track keywords
await track_keywords(client, config.keywords, config.domain)
report["credits_used"] += len(config.keywords)
# Scan Reddit
mentions = await scan_reddit(client, config.reddit_queries)
report["reddit_mentions"] = mentions
report["credits_used"] += len(config.reddit_queries)
cost = report["credits_used"] * 0.005
report["cost_usd"] = cost
print(f"Marketing Research Report - {report['date']}")
print(f"New competitor pages: {len(report['new_competitor_pages'])}")
print(f"New Reddit mentions: {len(report['reddit_mentions'])}")
print(f"Credits: {report['credits_used']} | Cost: {cost:.3f}")
for p in report["new_competitor_pages"]:
print(f" NEW [{p['competitor']}]: {p['title']}")
for m in report["reddit_mentions"]:
print(f" REDDIT [{m['query']}]: {m['title']}")
return report
asyncio.run(daily_research(config))Ejemplo en Python
import asyncio
import httpx
import sqlite3
from datetime import date
from pathlib import Path
SCAVIO_API_KEY = "your-api-key"
DB = Path("research.db")
async def main():
conn = sqlite3.connect(DB)
conn.execute("""CREATE TABLE IF NOT EXISTS keyword_ranks
(keyword TEXT, position INTEGER, checked_at TEXT)""")
keywords = ["best search API 2026", "cheap scraping API", "MCP search tool"]
today = date.today().isoformat()
async with httpx.AsyncClient(timeout=15) as client:
for kw in keywords:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": kw, "num_results": 10}
)
results = resp.json().get("results", [])
pos = next((i+1 for i, r in enumerate(results) if "scavio.dev" in r.get("url", "")), None)
conn.execute("INSERT INTO keyword_ranks VALUES (?,?,?)", (kw, pos, today))
status = f"#{pos}" if pos else "not found"
print(f" {kw}: {status}")
conn.commit()
cost = len(keywords) * 0.005
print(f"Credits: {len(keywords)} | Cost: {cost:.3f}")
conn.close()
asyncio.run(main())Ejemplo en JavaScript
const SCAVIO_API_KEY = "your-api-key";
const KEYWORDS = ["best search API 2026", "cheap scraping API", "MCP search tool"];
const TARGET = "scavio.dev";
async function trackKeyword(keyword) {
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": SCAVIO_API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ query: keyword, num_results: 10 })
});
const data = await resp.json();
const results = data.results || [];
const idx = results.findIndex(r => (r.url || "").includes(TARGET));
return { keyword, position: idx >= 0 ? idx + 1 : null };
}
async function main() {
const rankings = [];
for (const kw of KEYWORDS) {
const rank = await trackKeyword(kw);
rankings.push(rank);
console.log(` ${kw}: ${rank.position ? "#" + rank.position : "not found"}`);
}
console.log(`Credits: ${KEYWORDS.length} | Cost: $${(KEYWORDS.length * 0.005).toFixed(3)}`);
}
main();Salida esperada
Marketing Research Report - 2026-05-17
New competitor pages: 4
New Reddit mentions: 7
Credits: 16 | Cost: $0.080