dguiducci/skald-connectors
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

google-trends: fix broken MCP handshake and RSS articles (v2 / 1.1.0)

Browse Source 
The server never implemented `initialize`, so every MCP client got
-32601 to its opening request and aborted before listing a tool. Also
missing: `notifications/initialized` and `ping`; `tools/list` returned a
bare array instead of {"tools": [...]}; notifications (no `id`) got a
full response written to stdout.

Rewritten to the shape the other local connectors already use (TOOLS
manifest + TOOL_DISPATCH, `_text_result` with isError, handle_request
returning None for notifications).

Also fixed:
- include_articles always returned nothing: trendspyg emits
  `news_articles`, the mapper read `articles`. explore_link kept too.
- fn(**arguments) turned a bad argument into -32603; handlers now take
  an args dict and coerce/clamp.
- errors were returned as successful results; now isError: true, with
  trendspyg's typed exceptions translated into actionable messages.
- browser calls could run ~100s (10 retries x 8s); capped at ~25s.
- verify.py shipped but connector.json declared no `verify` block; wired
  it and upgraded the script to a real RSS fetch.
- explore mutated trendspyg's ExploreEnvelope; dropped the no-op
  output_format param; trendspyg pinned >=1.6.0 and imported defensively.
This commit is contained in:
Daniele
2026-08-24 17:29:36 +01:00
parent 0c04ba2e4c
commit 5e4c41183c
7 changed files with 401 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connectors/google-trends/google_trends_mcp.py
+319 -282
View File
@@ -2,14 +2,14 @@
"""Google Trends MCP server (JSON-RPC 2.0 over stdio) using trendspyg.
Capabilities (callable as `mcp__google-trends__<tool>`):
status — self-check: confirms trendspyg is importable
get_trending — current trending topics per country (fast, RSS, no browser)
get_interest_over_time — keyword interest history (browser-based, ~10-30s)
explore — full keyword analysis: interest over time + related queries
+ interest by region (browser-based, ~10-30s)
status — self-check: confirms trendspyg is importable
get_trending — current trending topics per country (fast, RSS, no browser)
get_interest_over_time — keyword interest history (browser-based, ~10-25s)
explore — full keyword analysis: interest over time + related
queries + interest by region (browser-based, ~10-25s)
Data source: trendspyg v0.7.0 (MIT) — Google Trends scraping library.
RSS path is lightweight (HTTP only); Explore path requires Chrome/Selenium.
Data source: trendspyg (MIT) — Google Trends scraping library.
The RSS path is lightweight (HTTP only); the Explore paths drive Chrome/Selenium.
Run with:
python3 google_trends_mcp.py
@@ -21,46 +21,87 @@ import json
import sys
from typing import Any
import trendspyg
# ── Logging ─────────────────────────────────────────────────────────────────────
def log(msg: str) -> None:
print(f"[google_trends_mcp] {msg}", file=sys.stderr, flush=True)
# ── trendspyg import ────────────────────────────────────────────────────────────
# Imported at module load, but a failure must not kill the server: the MCP
# handshake still has to succeed so skald can surface a readable tool error
# instead of a process that dies on startup.
try:
import trendspyg
_IMPORT_ERROR: str | None = None
except Exception as e: # pragma: no cover - depends on the install
trendspyg = None # type: ignore[assignment]
_IMPORT_ERROR = str(e)
log(f"trendspyg is not importable: {e}")
# Google's Explore pages soft-throttle; trendspyg retries past it. Its defaults
# (10 attempts × 8s) allow a ~100s call, far beyond an agent's patience — cap
# the worst case at roughly 25s instead.
MAX_RETRIES = 3
RETRY_WAIT = 6.0
def _describe_error(e: Exception) -> str:
"""Turn a trendspyg exception into a message an agent can act on."""
name = type(e).__name__
if name == "RateLimitError":
return f"Error: Google Trends is throttling this IP ({e}). Retry in a few minutes."
if name == "BrowserError":
return f"Error: Chrome could not be started ({e}). This tool needs Chrome/Selenium installed."
if name == "InvalidParameterError":
return f"Error: Invalid parameter: {e}"
return f"Error: Google Trends request failed ({name}): {e}"
# ── Argument helpers ────────────────────────────────────────────────────────────
def _str_arg(args: dict[str, Any], key: str, default: str = "") -> str:
value = args.get(key, default)
return value.strip() if isinstance(value, str) else default
def _int_arg(args: dict[str, Any], key: str, default: int, lo: int, hi: int) -> int:
try:
value = int(args.get(key, default))
except (TypeError, ValueError):
return default
return max(lo, min(hi, value))
def _bool_arg(args: dict[str, Any], key: str, default: bool) -> bool:
value = args.get(key, default)
return value if isinstance(value, bool) else default
# ── Tool implementations ────────────────────────────────────────────────────────
def _status() -> dict[str, Any]:
"""Self-check: verify trendspyg is importable and working."""
v = trendspyg.__version__
return {
"status": "ok",
"library": "trendspyg",
"version": v,
}
def _status(args: dict[str, Any]) -> str:
"""Self-check: confirm trendspyg is importable."""
if trendspyg is None:
return f"Error: trendspyg is not installed: {_IMPORT_ERROR}"
version = getattr(trendspyg, "__version__", "unknown")
return (
f"OK: Google Trends integration is operational (trendspyg v{version}).\n"
" - get_trending: ready (RSS, no browser needed)\n"
" - get_interest_over_time / explore: require Chrome/Selenium"
)
def _get_trending(
geo: str = "US",
max_trends: int = 10,
include_articles: bool = False,
) -> dict[str, Any]:
"""Get current trending search topics for a country (RSS — fast, no browser).
def _get_trending(args: dict[str, Any]) -> str:
"""Current trending search topics for a country (RSS — fast, no browser)."""
if trendspyg is None:
return f"Error: trendspyg is not installed: {_IMPORT_ERROR}"
Args:
geo: Two-letter country code (US, GB, IT, DE, JP, etc.).
max_trends: Max trending topics to return (default 10, max 20).
include_articles: Include news article headlines/URLs (default False).
Returns:
A dict with: geo, fetched_at, trending_trends[] (each with title,
traffic_volume, articles[] when requested).
"""
max_trends = min(max_trends, 20)
if max_trends < 1:
max_trends = 10
geo = _str_arg(args, "geo", "US") or "US"
max_trends = _int_arg(args, "max_trends", 10, 1, 20)
include_articles = _bool_arg(args, "include_articles", False)
try:
raw = trendspyg.download_google_trends_rss(
@@ -71,314 +112,310 @@ def _get_trending(
max_articles_per_trend=3,
)
except Exception as e:
return {"status": "error", "message": str(e), "geo": geo}
return _describe_error(e)
if not raw:
return {"status": "ok", "geo": geo, "trends": []}
# raw is a list of trend dicts
trends = []
for t in raw[:max_trends]:
for t in (raw or [])[:max_trends]:
item = {
"title": t.get("trend", ""),
"traffic": t.get("traffic", ""),
"traffic_min": t.get("traffic_min"),
"published": t.get("published", ""),
"explore_link": t.get("explore_link", ""),
}
if include_articles and "articles" in t:
if include_articles:
item["articles"] = [
{
"headline": a.get("headline", a.get("title", "")),
"headline": a.get("headline", ""),
"url": a.get("url", ""),
"source": a.get("source", ""),
}
for a in (t.get("articles") or [])[:3]
for a in (t.get("news_articles") or [])[:3]
]
trends.append(item)
return {"status": "ok", "geo": geo, "trends": trends}
return _json({"geo": geo, "count": len(trends), "trends": trends})
def _get_interest_over_time(
keyword: str,
geo: str = "US",
timeframe: str = "today 12-m",
output_format: str = "dict",
) -> dict[str, Any]:
"""Get a keyword's search interest over time (0-100 scale).
def _get_interest_over_time(args: dict[str, Any]) -> str:
"""A keyword's search interest over time (0-100 scale). Browser-based."""
if trendspyg is None:
return f"Error: trendspyg is not installed: {_IMPORT_ERROR}"
Browser-based — takes ~10-30s. Returns data points with date, value,
and is_partial flag.
keyword = _str_arg(args, "keyword")
if not keyword:
return "Error: Missing required parameter 'keyword'."
Args:
keyword: Search term to analyze.
geo: Two-letter country code (US, GB, IT, etc.) or '' for worldwide.
timeframe: Google Trends range ('today 12-m', 'today 5-y',
'now 7-d', 'now 1-H', 'all', or 'YYYY-MM-DD YYYY-MM-DD').
output_format: 'dict' (default) or 'json'.
Returns:
Dict with keyword, geo, timeframe, datapoints[].
"""
if not keyword or not keyword.strip():
return {"status": "error", "message": "keyword is required"}
geo = _str_arg(args, "geo", "US")
timeframe = _str_arg(args, "timeframe", "today 12-m") or "today 12-m"
try:
of = "dict" if output_format == "dict" else "json"
data = trendspyg.download_google_trends_interest_over_time(
keyword=keyword.strip(),
points = trendspyg.download_google_trends_interest_over_time(
keyword=keyword,
geo=geo,
timeframe=timeframe,
output_format=of,
output_format="dict",
max_retries=MAX_RETRIES,
retry_wait=RETRY_WAIT,
)
except Exception as e:
return {"status": "error", "message": str(e), "keyword": keyword}
return _describe_error(e)
if of == "json":
data = json.loads(data) if isinstance(data, str) else data
return {
"status": "ok",
points = points or []
return _json({
"keyword": keyword,
"geo": geo,
"timeframe": timeframe,
"datapoints": data,
}
"count": len(points),
"datapoints": points,
})
def _explore(
keyword: str,
geo: str = "US",
timeframe: str = "today 12-m",
include_related: bool = True,
include_geo: bool = True,
) -> dict[str, Any]:
"""Full keyword analysis: interest over time + related queries + region map.
def _explore(args: dict[str, Any]) -> str:
"""Full keyword analysis: interest over time + related queries + region map."""
if trendspyg is None:
return f"Error: trendspyg is not installed: {_IMPORT_ERROR}"
Browser-based — takes ~10-30s. Returns the complete Explore data in
one call.
Args:
keyword: Search term to analyze.
geo: Two-letter country code (US, GB, IT, etc.) or '' for worldwide.
timeframe: Google Trends range ('today 12-m', 'today 5-y', etc.).
include_related: Include related queries (top + rising).
include_geo: Include interest by region.
Returns:
Complete ExploreEnvelope as a JSON-safe dict.
"""
if not keyword or not keyword.strip():
return {"status": "error", "message": "keyword is required"}
keyword = _str_arg(args, "keyword")
if not keyword:
return "Error: Missing required parameter 'keyword'."
try:
data = trendspyg.download_google_trends_explore(
keyword=keyword.strip(),
geo=geo,
timeframe=timeframe,
include_related=include_related,
include_geo=include_geo,
envelope = trendspyg.download_google_trends_explore(
keyword=keyword,
geo=_str_arg(args, "geo", "US"),
timeframe=_str_arg(args, "timeframe", "today 12-m") or "today 12-m",
include_related=_bool_arg(args, "include_related", True),
include_geo=_bool_arg(args, "include_geo", True),
max_retries=MAX_RETRIES,
retry_wait=RETRY_WAIT,
)
except Exception as e:
return {"status": "error", "message": str(e), "keyword": keyword}
return _describe_error(e)
# data is already a dict (ExploreEnvelope)
data["status"] = "ok"
return data
# trendspyg returns an ExploreEnvelope: already JSON-safe, every field present.
return _json(envelope)
# ── JSON-RPC helpers ───────────────────────────────────────────────────────────
def _json(payload: Any) -> str:
return json.dumps(payload, ensure_ascii=False, default=str)
TOOLS: dict[str, tuple[Any, dict[str, Any]]] = {
"status": (
_status,
{
"description": "Check if Google Trends MCP is operational",
"inputSchema": {"type": "object", "properties": {}, "required": []},
},
),
"get_trending": (
_get_trending,
{
"description": "Get current trending search topics for a country (RSS — fast, no browser)",
"inputSchema": {
"type": "object",
"properties": {
"geo": {
"type": "string",
"description": "Two-letter country code (US, GB, IT, DE, etc.)",
"default": "US",
},
"max_trends": {
"type": "number",
"description": "Max trending topics (1-20)",
"default": 10,
},
"include_articles": {
"type": "boolean",
"description": "Include news article headlines/URLs",
"default": False,
},
# ── Tool manifest ────────────────────────────────────────────────────────────────
TOOLS = [
{
"name": "status",
"title": "Status Check",
"description": (
"Self-check that the Google Trends integration is operational: confirms "
"the trendspyg library is installed and reports its version. "
"Call this first if any Google Trends tool fails."
),
"inputSchema": {"type": "object", "properties": {}},
},
{
"name": "get_trending",
"title": "Get Trending Topics",
"description": (
"Get the search topics trending RIGHT NOW in a country, with their traffic "
"volume and (optionally) the news articles behind them. Fast (~1s, no browser). "
"Always current — it cannot look at past dates or filter by keyword; "
"use get_interest_over_time for a specific term."
),
"inputSchema": {
"type": "object",
"properties": {
"geo": {
"type": "string",
"description": "Country or region code (e.g. 'US', 'GB', 'IT', 'US-CA'). Default: 'US'.",
},
"required": [],
},
},
),
"get_interest_over_time": (
_get_interest_over_time,
{
"description": "Get a keyword's search interest history (0-100 scale, browser-based ~10-30s)",
"inputSchema": {
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "Search term to analyze",
},
"geo": {
"type": "string",
"description": "Two-letter country code or '' for worldwide",
"default": "US",
},
"timeframe": {
"type": "string",
"description": "Google Trends range: 'today 12-m', 'today 5-y', 'now 7-d', 'all'",
"default": "today 12-m",
},
"output_format": {
"type": "string",
"description": "Output format: 'dict' or 'json'",
"default": "dict",
},
"max_trends": {
"type": "integer",
"description": "Maximum number of trending topics to return (1-20). Default: 10.",
},
"required": ["keyword"],
},
},
),
"explore": (
_explore,
{
"description": "Full keyword analysis: interest over time + related queries + region map (browser-based ~10-30s)",
"inputSchema": {
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "Search term to analyze",
},
"geo": {
"type": "string",
"description": "Two-letter country code or '' for worldwide",
"default": "US",
},
"timeframe": {
"type": "string",
"description": "Google Trends range: 'today 12-m', 'today 5-y', 'now 7-d', 'all'",
"default": "today 12-m",
},
"include_related": {
"type": "boolean",
"description": "Include top + rising related queries",
"default": True,
},
"include_geo": {
"type": "boolean",
"description": "Include interest by region",
"default": True,
},
"include_articles": {
"type": "boolean",
"description": "Include up to 3 news article headlines/URLs per trend. Default: false.",
},
"required": ["keyword"],
},
"required": [],
},
),
},
{
"name": "get_interest_over_time",
"title": "Interest Over Time",
"description": (
"Get a keyword's search interest history as Google's 0-100 relative index, "
"oldest point first. Use it to see how a term's popularity moved over a period. "
"Browser-based: takes ~10-25s and needs Chrome. A keyword Google has no data "
"for comes back as a series of zeros."
),
"inputSchema": {
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "Search term to analyze (e.g. 'bitcoin').",
},
"geo": {
"type": "string",
"description": "Country or region code (e.g. 'US', 'IT', 'US-CA'), or '' for worldwide. Default: 'US'.",
},
"timeframe": {
"type": "string",
"description": (
"Google Trends date range: 'today 12-m' (default), 'today 5-y', "
"'today 3-m', 'now 7-d', 'now 1-H', 'all', or a custom "
"'YYYY-MM-DD YYYY-MM-DD'."
),
},
},
"required": ["keyword"],
},
},
{
"name": "explore",
"title": "Explore Keyword",
"description": (
"Full Google Trends Explore picture for one keyword in a single call: interest "
"over time, related queries (top + rising), and interest by region. Prefer this "
"over get_interest_over_time when you also want related terms or the regional "
"breakdown. Browser-based: takes ~10-25s and needs Chrome."
),
"inputSchema": {
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "Search term to analyze (e.g. 'bitcoin').",
},
"geo": {
"type": "string",
"description": "Country or region code (e.g. 'US', 'IT', 'US-CA'), or '' for worldwide. Default: 'US'.",
},
"timeframe": {
"type": "string",
"description": (
"Google Trends date range: 'today 12-m' (default), 'today 5-y', "
"'now 7-d', 'all', or a custom 'YYYY-MM-DD YYYY-MM-DD'."
),
},
"include_related": {
"type": "boolean",
"description": "Include top + rising related queries. Default: true.",
},
"include_geo": {
"type": "boolean",
"description": "Include interest by region. Default: true.",
},
},
"required": ["keyword"],
},
},
]
TOOL_DISPATCH = {
"status": _status,
"get_trending": _get_trending,
"get_interest_over_time": _get_interest_over_time,
"explore": _explore,
}
# ── JSON-RPC request handler ───────────────────────────────────────────────────
# ── JSON-RPC dispatch ────────────────────────────────────────────────────────────
def handle_request(request: dict[str, Any]) -> dict[str, Any]:
"""Dispatch a JSON-RPC 2.0 request and return a response."""
req_id = request.get("id", None)
method = request.get("method", "")
params = request.get("params", {})
if method == "tools/list":
tools_list = []
for name, (_, schema) in TOOLS.items():
t = {
"name": name,
"description": schema["description"],
"inputSchema": schema["inputSchema"],
}
# Set title for friendly name in UI
title_map = {
"status": "Status Check",
"get_trending": "Get Trending Topics",
"get_interest_over_time": "Interest Over Time",
"explore": "Explore Keyword",
}
if name in title_map:
t["title"] = title_map[name]
tools_list.append(t)
return _ok(req_id, tools_list)
elif method == "tools/call":
tool_name = params.get("name", "")
arguments = params.get("arguments", {})
if tool_name not in TOOLS:
return _error(req_id, -32601, f"Method not found: {tool_name}")
try:
fn, _ = TOOLS[tool_name]
result = fn(**arguments)
# Ensure result is JSON-serializable and wrapped in content
if isinstance(result, dict):
content = [{"type": "text", "text": json.dumps(result, default=str)}]
else:
content = [{"type": "text", "text": str(result)}]
return {"jsonrpc": "2.0", "id": req_id, "result": {"content": content}}
except Exception as e:
log(f"Error calling {tool_name}: {e}")
return _error(req_id, -32603, f"Internal error: {e}")
else:
return _error(req_id, -32601, f"Method not found: {method}")
def _ok(req_id: Any, result: Any) -> str:
return json.dumps({"jsonrpc": "2.0", "id": req_id, "result": result})
def _ok(req_id: Any, result: Any) -> dict[str, Any]:
return {"jsonrpc": "2.0", "id": req_id, "result": result}
def _text_result(req_id: Any, text: str, is_error: bool = False) -> str:
payload: dict = {
"jsonrpc": "2.0",
"id": req_id,
"result": {"content": [{"type": "text", "text": text}]},
}
if is_error:
payload["result"]["isError"] = True
return json.dumps(payload)
def _error(req_id: Any, code: int, message: str) -> dict[str, Any]:
return {
def _error(req_id: Any, code: int, message: str) -> str:
return json.dumps({
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": code, "message": message},
}
})
# ── Main loop ──────────────────────────────────────────────────────────────────
def handle_request(msg: dict) -> str | None:
method = msg.get("method", "")
req_id = msg.get("id")
if method == "initialize":
return _ok(req_id, {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "google-trends",
"version": "1.1.0",
},
})
if method == "notifications/initialized":
return None
if method == "ping":
return _ok(req_id, {})
if method == "tools/list":
return _ok(req_id, {"tools": TOOLS})
if method == "tools/call":
params = msg.get("params", {})
tool_name = params.get("name", "")
tool_args = params.get("arguments") or {}
handler = TOOL_DISPATCH.get(tool_name)
if handler is None:
return _text_result(req_id, f"Error: Unknown tool: {tool_name}", is_error=True)
try:
result = handler(tool_args)
is_err = result.startswith("Error:")
return _text_result(req_id, result, is_error=is_err)
except Exception as e:
log(f"Unhandled exception in tool '{tool_name}': {e}")
return _text_result(req_id, f"Error: Internal error in tool '{tool_name}': {e}", is_error=True)
# A notification we do not handle gets no reply — only requests do.
if req_id is None:
return None
return _error(req_id, -32601, f"Method not found: {method}")
# ── Main loop ────────────────────────────────────────────────────────────────────
def main() -> None:
"""Read JSON-RPC 2.0 requests from stdin and write responses to stdout."""
log("Google Trends MCP server starting...")
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
request = json.loads(line)
except json.JSONDecodeError as e:
log(f"Invalid JSON: {e}")
resp = _error(None, -32700, f"Parse error: {e}")
else:
resp = handle_request(request)
log("Starting Google Trends MCP server")
try:
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
msg = json.loads(line)
except json.JSONDecodeError as e:
log(f"Invalid JSON input: {e}")
continue
sys.stdout.write(json.dumps(resp) + "\n")
sys.stdout.flush()
log("Google Trends MCP server shutting down.")
resp = handle_request(msg)
if resp is not None:
sys.stdout.write(resp + "\n")
sys.stdout.flush()
except KeyboardInterrupt:
pass
if __name__ == "__main__":