Add real-time WebSocket tools

.env.example

@@ -12,3 +12,8 @@ NEWSDATA_API_KEY="your_newsdata_api_key_here"
 # NEWSDATA_MAX_RETRIES=5
 # NEWSDATA_RETRY_BACKOFF=2.0
 # NEWSDATA_RETRY_BACKOFF_MAX=60.0
+
+# Optional: real-time WebSocket settings, used by the `stream_news` tool.
+# NEWSDATA_WS_URL=wss://ws.newsdata.io/ws/event
+# Ceiling on stream_news's wait_seconds, in seconds (default: 120)
+# NEWSDATA_WS_MAX_WAIT=120

README.md

@@ -72,6 +72,8 @@ cp .env.example .env
 | `NEWSDATA_MAX_RETRIES` | `5` | Maximum attempts for transient failures (network, 5xx, 429). |
 | `NEWSDATA_RETRY_BACKOFF` | `2.0` | Base for exponential backoff (`base * 2^(attempt-1)`). Seconds. |
 | `NEWSDATA_RETRY_BACKOFF_MAX` | `60.0` | Cap on a single retry sleep, seconds. |
+| `NEWSDATA_WS_URL` | `wss://ws.newsdata.io/ws/event` | Real-time endpoint used by `stream_news`. |
+| `NEWSDATA_WS_MAX_WAIT` | `120` | Ceiling on `stream_news`'s `wait_seconds`, in seconds. |
 | `NEWSDATA_INTEGRATION_KEY` | _(unset)_ | Used only by `pytest -m integration`. Without it, live-API tests skip. |
 
 All values are read at module import time; restart the server after changing them.
@@ -160,8 +162,75 @@ Then in **ChatGPT → Settings → Connectors → Add custom connector**, regist
 | `get_news_counts` | `/api/1/count` | Aggregate article counts over a date range (`hour` / `day` buckets or single `all` total) |
 | `get_crypto_counts` | `/api/1/crypto/count` | Aggregate crypto article counts over a date range |
 | `get_market_counts` | `/api/1/market/count` | Aggregate market article counts over a date range |
+| `register_realtime_query` | `/api/1/websocket/register` | Register a standing real-time query, returns a `registration_id` |
+| `list_realtime_queries` | `/api/1/websocket/fetch` | List the account's registered real-time queries |
+| `delete_realtime_query` | `/api/1/websocket/delete` | Delete a registered real-time query |
+| `stream_news` | `wss://ws.newsdata.io/ws/event` | Collect real-time articles for a registered query ([below](#real-time-news-websocket)) |
 
-All tools are read-only and idempotent; the MCP-protocol annotations let compatible clients (Claude Code, MCP Inspector, etc.) cache and parallelize calls.
+The eight REST news tools are read-only and idempotent; the MCP-protocol annotations let compatible clients (Claude Code, MCP Inspector, etc.) cache and parallelize calls. `register_realtime_query` and `delete_realtime_query` are annotated as mutating (the latter destructive), since they create and remove server-side state.
+
+---
+
+## Real-time news (WebSocket)
+
+Real time is a **two-step flow**: register a standing query once, then collect
+from it as often as you like.
+
+```text
+register_realtime_query(
+  q="bitcoin",
+  language="en"
+)
+→ registration_id: a1b2c3d4e5f6
+```
+
+```text
+stream_news(
+  registration_id="a1b2c3d4e5f6",
+  max_articles=10,
+  wait_seconds=30
+)
+```
+
+`register_realtime_query` takes the familiar filter parameters (`q`, `country`,
+`language`, `domain`, …). There are **no date or paging filters** — a registered
+query matches news as it is published. Registering identical filters twice
+answers `Error (HTTP 409)`; call `list_realtime_queries` first to recover the
+existing id.
+
+### Why `stream_news` is bounded
+
+MCP tools are request/response, so an open-ended stream cannot be a tool.
+`stream_news` listens on a live connection and returns as soon as **either**
+`max_articles` have arrived **or** `wait_seconds` elapses — whichever comes
+first. It always returns within `wait_seconds`.
+
+Every reply states why it stopped, so the model can act on it:
+
+| `stopped_because` | Meaning |
+|---|---|
+| `max_articles` | Hit the cap — there may be more waiting |
+| `timeout` | The window elapsed; a `0` count means the feed was simply quiet |
+| `connection_closed` | The server ended the feed; returns whatever was collected |
+
+Call it again to keep listening — the registration stays alive between calls.
+An empty result is normal for a narrow query and does **not** mean anything
+failed.
+
+Bounds: `max_articles` is clamped to 1–50, `wait_seconds` to 1–`NEWSDATA_WS_MAX_WAIT`
+(120 by default).
+
+The server always accepts the handshake and then closes with code **1008** when
+the connection is refused, carrying one of three reasons: `invalid credentials
+or registration not found`, `api limit reached`, or `device limit reached` (more
+than 5 devices on one `registration_id`). Those return an `Error` and are not
+retried. Every other close code — including `1013` (`send timeout`) — is
+transient and surfaces as `stopped_because: connection_closed` with whatever was
+collected.
+
+> **Cost.** Each delivered article consumes **1 API credit per connected
+> device**. A broad query over a long window can burn credits quickly — keep
+> `max_articles` no higher than you need.
 
 ---
 
@@ -247,6 +316,8 @@ Notes on parameter shapes:
 - Free plan results are delayed relative to paid plans.
 - Result `size` is capped by plan tier: commonly 10 results on free, up to 50 on paid plans.
 - The count endpoints return aggregate buckets (one per `interval` slot) rather than article content.
+- Real-time coverage needs a plan with WebSocket access; without it `stream_news` returns `Error: invalid credentials or registration not found`.
+- Real-time articles are billed per delivered article, per connected device.
 - Every tool returns plain text (the MCP-protocol return type). Errors come back as `Error (HTTP 4xx): …` with the status code and a friendly message; HTTP 429 errors include a `retry after Ns` hint when the upstream `Retry-After` header was parseable.
 
 Full API reference: [https://newsdata.io/documentation](https://newsdata.io/documentation). Machine-readable contract: [OpenAPI 3.1 spec](https://newsdata.io/openapi.json).

pyproject.toml

@@ -37,6 +37,7 @@ dependencies = [
     "mcp[cli]>=1.27.0,<2",
     "pydantic>=2.7.0,<3",
     "python-dotenv>=1.2.2,<2",
+    "websockets>=16,<17",
 ]
 
 [project.scripts]

src/newsdata_mcp/_mcp.py

@@ -24,6 +24,25 @@
     openWorldHint=True,
 )
 
+# Registering a real-time query creates server-side state. It is not
+# destructive (nothing is lost) and is idempotent in practice — a second
+# identical registration answers 409 with the existing id rather than
+# creating a duplicate.
+CREATE_TOOL = ToolAnnotations(
+    readOnlyHint=False,
+    destructiveHint=False,
+    idempotentHint=True,
+    openWorldHint=True,
+)
+
+# Deleting a registered query removes server-side state for good.
+DESTRUCTIVE_TOOL = ToolAnnotations(
+    readOnlyHint=False,
+    destructiveHint=True,
+    idempotentHint=True,
+    openWorldHint=True,
+)
+
 
 @asynccontextmanager
 async def _lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:

src/newsdata_mcp/formatters.py

@@ -255,3 +255,91 @@ def format_sources(data: dict[str, Any]) -> str:
         lines.append("")
 
     return "\n".join(lines)
+
+
+def format_registration(data: dict[str, Any]) -> str:
+    """Render a `websocket/register` response: the new query's
+    ``registration_id`` plus whatever else the server echoes back.
+    """
+    if data.get("status") != "success":
+        return _format_error(data)
+
+    results = (data.get("data", {}) or {}).get("results") or {}
+    if not isinstance(results, dict):
+        return "Registered, but the response carried no registration details."
+
+    lines = ["registered: yes"]
+    _append_field(lines, "registration_id", results.get("registration_id"))
+    for key, value in results.items():
+        if key == "registration_id":
+            continue
+        _append_field(lines, key, value)
+    return "\n".join(lines)
+
+
+def format_registered_queries(data: dict[str, Any]) -> str:
+    """Render a `websocket/fetch` response: one block per registered
+    real-time query.
+    """
+    if data.get("status") != "success":
+        return _format_error(data)
+
+    results = (data.get("data", {}) or {}).get("results") or {}
+    queries = results.get("queries") if isinstance(results, dict) else None
+    if not queries:
+        return "No real-time queries are registered on this account."
+
+    lines = [f"registered_queries: {len(queries)}", ""]
+    for index, query in enumerate(queries, 1):
+        lines.append(f"Query {index}:")
+        if isinstance(query, dict):
+            _append_field(lines, "registration_id", query.get("registration_id"))
+            for key, value in query.items():
+                if key == "registration_id":
+                    continue
+                _append_field(lines, key, value)
+        else:
+            _append_field(lines, "value", query)
+        lines.append("")
+    return "\n".join(lines)
+
+
+def format_deletion(data: dict[str, Any], registration_id: str) -> str:
+    """Render a `websocket/delete` response."""
+    if data.get("status") != "success":
+        return _format_error(data)
+    return f"deleted: yes\nregistration_id: {registration_id}"
+
+
+def format_stream(
+    articles: list[dict[str, Any]],
+    registration_id: str,
+    stopped_because: str,
+    waited_seconds: float,
+) -> str:
+    """Render the articles collected by a bounded `stream_news` call.
+
+    ``stopped_because`` explains why collection ended (``max_articles``,
+    ``timeout``, or ``connection_closed``) so the model can tell "the
+    feed is quiet" from "I hit the cap and there may be more".
+    """
+    header = [
+        f"registration_id: {registration_id}",
+        f"collected_articles: {len(articles)}",
+        f"stopped_because: {stopped_because}",
+        f"waited_seconds: {round(waited_seconds, 1)}",
+    ]
+    if not articles:
+        header.append("")
+        header.append(
+            "No matching articles were published during this window. "
+            "The query is still registered — call again to keep listening."
+        )
+        return "\n".join(header)
+
+    lines = [*header, ""]
+    for index, article in enumerate(articles, 1):
+        lines.append(f"Article {index}:")
+        lines.extend(_format_article_item(article))
+        lines.append("")
+    return "\n".join(lines)

src/newsdata_mcp/http.py

@@ -179,7 +179,7 @@ async def close_client() -> None:
 
 
 async def _request_once(
-    endpoint: str, clean: dict[str, Any]
+    endpoint: str, clean: dict[str, Any], method: str = "GET"
 ) -> tuple[dict[str, Any], bool]:
     """Execute one HTTP attempt.
 
@@ -191,7 +191,8 @@ async def _request_once(
     response: httpx.Response | None = None
     try:
         client = await _get_client()
-        response = await client.get(
+        response = await client.request(
+            method,
             f"{NEWSDATA_BASE_URL}/{endpoint}",
             params=clean,
         )
@@ -307,7 +308,9 @@ async def _request_once(
         )
 
 
-async def fetch(endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
+async def fetch(
+    endpoint: str, params: dict[str, Any], method: str = "GET"
+) -> dict[str, Any]:
     """Public entry point: one logical request, with retries.
 
     Permanent failures (auth, validation, soft errors, JSON decode,
@@ -324,15 +327,15 @@ async def fetch(endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
         }
 
     clean = _normalize_params(params)
-    logger.info("Newsdata.io GET /%s (%d params)", endpoint, len(clean))
+    logger.info("Newsdata.io %s /%s (%d params)", method, endpoint, len(clean))
 
     last_envelope: dict[str, Any] = {
         "status": "error",
         "message": "Retry loop exhausted unexpectedly.",
         "status_code": None,
     }
     for attempt in range(1, MAX_RETRIES + 1):
-        envelope, retryable = await _request_once(endpoint, clean)
+        envelope, retryable = await _request_once(endpoint, clean, method)
         if not retryable:
             return envelope
         last_envelope = envelope

src/newsdata_mcp/params.py

@@ -130,6 +130,20 @@ def _csv_pattern(*values: str) -> str:
     ),
 ]
 
+REGISTRATION_ID = Annotated[
+    str,
+    Field(
+        min_length=32,
+        max_length=32,
+        description=(
+            "The 32-character `registration_id` of a real-time query, as "
+            "returned by `register_realtime_query` or listed by "
+            "`list_realtime_queries`. Do not invent one — look it up."
+        ),
+        examples=["9b2d1e8a7c4f4b6e9d3a5c7e1f2a4b6c"],
+    ),
+]
+
 ARTICLE_IDS = Annotated[
     str | list[str],
     Field(

src/newsdata_mcp/settings.py

@@ -50,5 +50,26 @@
     )
     RETRY_BACKOFF_MAX = 60.0
 
+# Real-time WebSocket endpoint used by the `stream_news` tool.
+NEWSDATA_WS_URL = os.getenv("NEWSDATA_WS_URL", "wss://ws.newsdata.io/ws/event")
+
+# The feed a registered query matches against.
+WS_NEWS_TYPE = "latest"
+
+# Close code the server uses for a permanent connection rejection.
+WS_POLICY_VIOLATION = 1008
+
+# Hard ceilings for `stream_news`, which must return within a tool call.
+try:
+    WS_MAX_WAIT_SECONDS = float(os.getenv("NEWSDATA_WS_MAX_WAIT", "120"))
+except (ValueError, TypeError):
+    warnings.warn(
+        "NEWSDATA_WS_MAX_WAIT must be a number; falling back to 120",
+        stacklevel=2,
+    )
+    WS_MAX_WAIT_SECONDS = 120.0
+
+WS_MAX_ARTICLES = 50
+
 if not NEWSDATA_API_KEY:
     warnings.warn("NEWSDATA_API_KEY is not set", stacklevel=2)

src/newsdata_mcp/tools/__init__.py

@@ -14,6 +14,7 @@
     latest,
     market,
     market_count,
+    realtime,
     sources,
 )
 
@@ -25,5 +26,6 @@
     "latest",
     "market",
     "market_count",
+    "realtime",
     "sources",
 ]