""" viralcli MCP server — live short-form trend data as native agent tools. Single self-contained file, made to be downloaded and run: curl -O https://www.viralcli.com/viralcli_mcp.py pip install "mcp[cli]" requests export VIRALCLI_API_KEY=sk_live_... # only needed for discover/niche/analyze Claude Code: claude mcp add viralcli -e VIRALCLI_API_KEY=sk_live_... -- python /abs/path/viralcli_mcp.py Claude Desktop / any MCP client (mcpServers): { "viralcli": { "command": "python", "args": ["/abs/path/viralcli_mcp.py"], "env": { "VIRALCLI_API_KEY": "sk_live_..." } } } Public tools (no key, no quota): live_trends, trending_sounds, rising_creators, search_insights, shop_trending, videos_momentum. Keyed tools (consume plan quota): discover, niche, analyze. Docs: https://www.viralcli.com/docs · Facts for agents: https://www.viralcli.com/llms.txt """ import os import time try: import requests except ImportError: # pragma: no cover raise SystemExit("needs `requests` (pip install requests)") from mcp.server.fastmcp import FastMCP BASE = os.environ.get("VIRALCLI_API_URL", "https://api.viralcli.com").rstrip("/") KEY = os.environ.get("VIRALCLI_API_KEY", "") _session = requests.Session() def _headers(): h = {"content-type": "application/json"} if KEY: h["Authorization"] = f"Bearer {KEY}" return h def _unwrap(r): if r.status_code >= 400: try: return {"error": r.status_code, "detail": r.json()} except ValueError: return {"error": r.status_code, "detail": r.text[:300]} j = r.json() return j.get("data", j) def _get(path, **params): return _unwrap(_session.get(BASE + path, headers=_headers(), params=params or None, timeout=60)) def _post(path, body): return _unwrap(_session.post(BASE + path, headers=_headers(), json=body, timeout=60)) def _await_job(sub, wait_seconds, poll=3): """Async endpoints return {job_id}; poll until done and return the result.""" job_id = isinstance(sub, dict) and (sub.get("job_id") or sub.get("id")) if not job_id: return sub deadline = time.time() + wait_seconds while time.time() < deadline: j = _get(f"/v1/jobs/{job_id}") if isinstance(j, dict) and j.get("status") in ("done", "failed", "error"): return j.get("result", j) time.sleep(poll) return {"status": "pending", "job_id": job_id} mcp = FastMCP("viralcli") # -- public (no key) ---------------------------------------------------------- @mcp.tool() def live_trends() -> dict: """Current curated + auto-discovered short-form trends: per-topic, per- platform (TikTok / Reels / Shorts) breakouts ranked by views/day velocity, HEATING/STEADY/SATURATING verdicts, sounds, seasonality, and 24h verdict shifts. Public — no key needed.""" return _get("/v1/trends") @mcp.tool() def trending_sounds() -> dict: """Named audio tracks currently carrying multiple breakout videos, ranked by the summed velocity of the distinct videos riding each sound. Public.""" return _get("/v1/sounds/trending") @mcp.tool() def rising_creators() -> dict: """Creators appearing repeatedly in current breakouts (2+ videos), ranked by summed views/day — the accounts producing winners right now. Public.""" return _get("/v1/creators/rising") @mcp.tool() def search_insights() -> dict: """Search-demand intelligence: real queries people search, in three views — all, trending (demand + healthy supply), and content_gap (high demand but thin supply = the openings worth making content for). Public.""" return _get("/v1/search/insights") @mcp.tool() def opportunity() -> dict: """"What should I make next?" as one transparent 0-100 Opportunity Score per topic: search DEMAND × how UNDER-SERVED it is × format MOMENTUM (heating verdict × measured velocity), plus a light seasonality nudge. Each item carries its component breakdown (demand/underserved/momentum/season), a coverage flag ('full' vs 'demand-only'), and the ranking's backtested confidence (walk-forward information coefficient). One ranked call to decide what to produce, grounded in this week's data. Public.""" return _get("/v1/opportunity") @mcp.tool() def shop_trending() -> dict: """TikTok Shop products ranked by SOLD-VELOCITY (units/day measured between snapshots of TikTok's own cumulative sold counts; cumulative sold until a product has 2+ snapshots). Each row: price, rating, reviews, seller, the search queries it rides, a sold-over-time series, and a TikTok Shop link. The commerce twin of video velocity — what's SELLING right now. Public.""" return _get("/v1/shop/trending") @mcp.tool() def videos_momentum() -> dict: """Videos ACCELERATING right now: tracked videos whose recent views/day (measured across watcher snapshots, last 48h) exceeds their lifetime pace — still speeding up, not just big. Includes a views sparkline per video. Trajectory data a single scrape cannot reconstruct. Public.""" return _get("/v1/videos/momentum") # -- keyed (consume plan quota) ------------------------------------------------ @mcp.tool() def discover(seed: str, platform: str = "youtube") -> dict: """Live discovery scrape for a keyword on one platform ("youtube" | "tiktok" | "instagram"): breakout videos ranked by velocity, an adoption histogram, and a heating/saturating verdict. Requires an API key; 1 discovery credit.""" return _post("/v1/discovery", {"seed": seed, "platform": platform}) @mcp.tool() def niche(seed: str, platform: str = "youtube", top_n: int = 5) -> dict: """A niche's most viral videos — each with its HOOK (the spoken opening that drives retention), stats, velocity, sound, and a storyboard URL. Search a niche like "marketing" or "programming" and get the proven openings to model content on. Requires an API key; charges top_n enrichment credits.""" sub = _post("/v1/niche", {"seed": seed, "platform": platform, "top_n": top_n, "density": 6}) return _await_job(sub, wait_seconds=300) @mcp.tool() def analyze(url: str) -> dict: """Analyze one TikTok / Reel / Short URL: stats, caption, hook transcript, storyboard URL, and views/day velocity. Requires an API key; 1 enrichment.""" sub = _post("/v1/analyze", {"url": url, "density": 9, "full": False}) return _await_job(sub, wait_seconds=180) if __name__ == "__main__": mcp.run()