init

src/opennews_mcp/__init__.py

@@ -0,0 +1 @@
+"""OpenNews MCP Server - Crypto news via 6551 REST/WebSocket API."""

src/opennews_mcp/__main__.py

@@ -0,0 +1,3 @@
+from opennews_mcp.server import main
+
+main()

src/opennews_mcp/api_client.py

@@ -0,0 +1,156 @@
+"""HTTP/WebSocket client for the 6551 news platform API."""
+
+import asyncio
+import json
+import time
+import logging
+from typing import Any, Optional
+
+import httpx
+
+from opennews_mcp.config import API_BASE_URL, WSS_URL, API_TOKEN
+
+logger = logging.getLogger(__name__)
+
+MAX_RETRIES = 2
+
+
+class NewsAPIClient:
+    """Async HTTP client for the 6551 news REST API."""
+
+    def __init__(self, base_url: str = API_BASE_URL, token: str = API_TOKEN):
+        self.base_url = base_url.rstrip("/")
+        self.token = token
+        self._client: Optional[httpx.AsyncClient] = None
+
+    def _headers(self) -> dict:
+        return {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json",
+        }
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            self._client = httpx.AsyncClient(
+                timeout=httpx.Timeout(30.0),
+                headers=self._headers(),
+            )
+        return self._client
+
+    async def _reset_client(self):
+        """Force close and recreate the HTTP client."""
+        if self._client and not self._client.is_closed:
+            await self._client.aclose()
+        self._client = None
+
+    async def close(self):
+        await self._reset_client()
+
+    # ---------- internal request with retry ----------
+
+    async def _request(self, method: str, url: str, **kwargs) -> httpx.Response:
+        """Execute an HTTP request with automatic retry on connection errors."""
+        last_exc = None
+        for attempt in range(MAX_RETRIES + 1):
+            try:
+                client = await self._get_client()
+                resp = await client.request(method, url, **kwargs)
+                resp.raise_for_status()
+                return resp
+            except (httpx.ConnectError, httpx.RemoteProtocolError) as e:
+                last_exc = e
+                logger.warning(
+                    "Connection error (attempt %d/%d): %s",
+                    attempt + 1, MAX_RETRIES + 1, repr(e),
+                )
+                await self._reset_client()
+            except httpx.HTTPStatusError:
+                raise
+        raise last_exc  # type: ignore[misc]
+
+    # ---------- REST endpoints ----------
+
+    async def get_engine_tree(self) -> dict:
+        """GET /open/news_type — 获取所有新闻源分类"""
+        resp = await self._request("GET", f"{self.base_url}/open/news_type")
+        return resp.json()
+
+    async def search_news(
+        self,
+        coins: Optional[list[str]] = None,
+        query: Optional[str] = None,
+        engine_types: Optional[dict[str, list[str]]] = None,
+        has_coin: bool = False,
+        limit: int = 20,
+        page: int = 1,
+    ) -> dict:
+        """POST /open/news_search — 搜索新闻文章"""
+        body: dict[str, Any] = {"limit": limit, "page": page}
+        if coins:
+            body["coins"] = coins
+        if query:
+            body["q"] = query
+        if engine_types:
+            body["engineTypes"] = engine_types
+        if has_coin:
+            body["hasCoin"] = has_coin
+
+        resp = await self._request("POST", f"{self.base_url}/open/news_search", json=body)
+        return resp.json()
+
+
+
+class NewsWSClient:
+    """WebSocket client for real-time news subscription."""
+
+    def __init__(self, wss_url: str = WSS_URL, token: str = API_TOKEN):
+        self.wss_url = f"{wss_url}?token={token}"
+        self._ws = None
+        self._request_id = 0
+
+    def _next_id(self) -> str:
+        self._request_id += 1
+        return f"req_{self._request_id}_{int(time.time())}"
+
+    async def connect(self):
+        import websockets
+        self._ws = await websockets.connect(self.wss_url)
+
+    async def close(self):
+        if self._ws:
+            await self._ws.close()
+            self._ws = None
+
+    async def subscribe_latest(
+        self,
+        engine_types: Optional[dict[str, list[str]]] = None,
+        coins: Optional[list[str]] = None,
+        has_coin: bool = False,
+    ) -> dict:
+        """订阅新闻推送，支持过滤器"""
+        if not self._ws:
+            await self.connect()
+        req_id = self._next_id()
+        params: dict[str, Any] = {}
+        if engine_types:
+            params["engineTypes"] = engine_types
+        if coins:
+            params["coins"] = coins
+        if has_coin:
+            params["hasCoin"] = has_coin
+        msg = {"method": "news.subscribe", "id": req_id, "params": params}
+        await self._ws.send(json.dumps(msg))
+        resp = await self._ws.recv()
+        return json.loads(resp)
+
+    async def receive_news(self, timeout: float = 10.0) -> Optional[dict]:
+        if not self._ws:
+            return None
+        try:
+            msg = await asyncio.wait_for(self._ws.recv(), timeout=timeout)
+            return json.loads(msg)
+        except asyncio.TimeoutError:
+            return None
+        except Exception as e:
+            logger.warning("WebSocket receive error: %s", repr(e))
+            return None

src/opennews_mcp/app.py

@@ -0,0 +1,54 @@
+"""FastMCP application instance, lifespan, and knowledge resources."""
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from pathlib import Path
+
+from mcp.server.fastmcp import FastMCP
+
+from opennews_mcp.api_client import NewsAPIClient, NewsWSClient
+
+# Knowledge directory (project root / knowledge)
+KNOWLEDGE_DIR = Path(__file__).resolve().parent.parent.parent / "knowledge"
+
+
+@dataclass
+class AppContext:
+    """Shared application state available to all tools via ctx."""
+    api: NewsAPIClient
+    ws: NewsWSClient
+
+
+@asynccontextmanager
+async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
+    """Manage the API client lifecycle."""
+    api = NewsAPIClient()
+    ws = NewsWSClient()
+    try:
+        yield AppContext(api=api, ws=ws)
+    finally:
+        await api.close()
+        await ws.close()
+
+
+# ---------- FastMCP instance ----------
+mcp = FastMCP(
+    "opennews-6551",
+    lifespan=app_lifespan,
+    json_response=True,
+)
+
+
+# ---------- Knowledge resources ----------
+def _read_knowledge(name: str) -> str:
+    path = KNOWLEDGE_DIR / name
+    if path.exists():
+        return path.read_text(encoding="utf-8")
+    return f"Knowledge file '{name}' not found."
+
+
+@mcp.resource("knowledge://guide")
+async def knowledge_guide() -> str:
+    """Usage guide — tool workflows, search strategies, best practices."""
+    return _read_knowledge("guide.md")

src/opennews_mcp/config.py

@@ -0,0 +1,57 @@
+"""Configuration and shared utilities for the OpenNews MCP server.
+
+Reads settings from config.json at project root. Environment variables
+can still override any value.
+"""
+
+import json
+import os
+from datetime import datetime, date
+from decimal import Decimal
+from pathlib import Path
+
+# ---------- Load config.json ----------
+_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+_CONFIG_PATH = _PROJECT_ROOT / "config.json"
+
+_cfg: dict = {}
+if _CONFIG_PATH.exists():
+    with open(_CONFIG_PATH, "r", encoding="utf-8") as f:
+        _cfg = json.load(f)
+
+# ---------- API (env vars take precedence) ----------
+API_BASE_URL = os.environ.get("OPENNEWS_API_BASE") or _cfg.get("api_base_url", "")
+WSS_URL      = os.environ.get("OPENNEWS_WSS_URL")  or _cfg.get("wss_url", "")
+API_TOKEN    = os.environ.get("OPENNEWS_TOKEN")    or _cfg.get("api_token", "")
+
+# 检查 token 是否配置
+if not API_TOKEN:
+    raise ValueError(
+        "OPENNEWS_TOKEN 未配置。请前往 https://6551.io/mcp 申请 API Token，"
+        "然后设置环境变量 OPENNEWS_TOKEN 或在 config.json 中配置 api_token。"
+    )
+
+# ---------- Safety ----------
+MAX_ROWS = int(os.environ.get("OPENNEWS_MAX_ROWS", 0) or _cfg.get("max_rows", 100))
+
+
+def clamp_limit(limit: int) -> int:
+    """Clamp user-supplied limit to [1, MAX_ROWS]."""
+    return min(max(1, limit), MAX_ROWS)
+
+
+def make_serializable(obj):
+    """Recursively convert non-JSON-serializable types."""
+    if obj is None:
+        return None
+    if isinstance(obj, dict):
+        return {k: make_serializable(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [make_serializable(item) for item in obj]
+    if isinstance(obj, (datetime, date)):
+        return obj.isoformat()
+    if isinstance(obj, Decimal):
+        return float(obj)
+    if isinstance(obj, bytes):
+        return obj.decode("utf-8", errors="replace")
+    return obj

src/opennews_mcp/server.py

@@ -0,0 +1,24 @@
+"""Entry point for the OpenNews MCP server."""
+
+import sys
+
+# psycopg3 async requires SelectorEventLoop on Windows (ProactorEventLoop is unsupported).
+if sys.platform == "win32":
+    import asyncio, selectors  # noqa: E401
+    asyncio.set_event_loop_policy(
+        asyncio.WindowsSelectorEventLoopPolicy()
+    )
+
+from opennews_mcp.app import mcp
+
+# Importing the tools package triggers registration of all @mcp.tool() decorators.
+import opennews_mcp.tools  # noqa: F401
+
+
+def main():
+    """Run the MCP server (stdio transport by default)."""
+    mcp.run()
+
+
+if __name__ == "__main__":
+    main()

src/opennews_mcp/tools/__init__.py

@@ -0,0 +1,8 @@
+"""Import all tool modules to register their @mcp.tool() decorators.
+
+Each module is self-contained and atomic — no inter-tool dependencies.
+"""
+
+from opennews_mcp.tools import discovery  # noqa: F401
+from opennews_mcp.tools import news       # noqa: F401
+from opennews_mcp.tools import realtime   # noqa: F401

src/opennews_mcp/tools/discovery.py

@@ -0,0 +1,79 @@
+"""Discovery tools — list available news sources and categories."""
+
+from mcp.server.fastmcp import Context
+
+from opennews_mcp.app import mcp
+
+
+@mcp.tool()
+async def get_news_sources(ctx: Context) -> dict:
+    """Get all available news source categories and their metadata.
+
+    Returns a tree structure with engine types (news, listing, onchain, meme, market)
+    and their sub-categories (Bloomberg, Reuters, Binance, etc.).
+
+    Use this first to understand what news sources are available before searching.
+    """
+    api = ctx.request_context.lifespan_context.api
+
+    try:
+        result = await api.get_engine_tree()
+        data = result.get("data", [])
+
+        # Build a simplified summary
+        sources = []
+        for engine in data:
+            categories = []
+            for cat in engine.get("categories", []):
+                categories.append({
+                    "code": cat.get("code"),
+                    "name": cat.get("name"),
+                    "enName": cat.get("enName"),
+                    "aiEnabled": cat.get("aiEnabled", False),
+                })
+            sources.append({
+                "code": engine.get("code"),
+                "name": engine.get("name"),
+                "enName": engine.get("enName"),
+                "category_count": len(categories),
+                "categories": categories,
+            })
+
+        return {
+            "success": True,
+            "data": sources,
+            "engine_count": len(sources),
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def list_news_types(ctx: Context) -> dict:
+    """List all available news type codes for filtering.
+
+    Returns a flat list of news source codes that can be used with
+    the newsType parameter in search_news.
+    """
+    api = ctx.request_context.lifespan_context.api
+
+    try:
+        result = await api.get_engine_tree()
+        data = result.get("data", [])
+
+        types = []
+        for engine in data:
+            for cat in engine.get("categories", []):
+                types.append({
+                    "code": cat.get("code"),
+                    "engineType": engine.get("code"),
+                    "name": cat.get("enName") or cat.get("name"),
+                })
+
+        return {
+            "success": True,
+            "data": types,
+            "count": len(types),
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}

src/opennews_mcp/tools/news.py

@@ -0,0 +1,227 @@
+"""News content tools — search and retrieve crypto news via REST API.
+
+Uses POST /open/news_search as the primary data source.
+Returns raw article data as-is from the API.
+"""
+
+from mcp.server.fastmcp import Context
+
+from opennews_mcp.app import mcp
+from opennews_mcp.config import clamp_limit, make_serializable, MAX_ROWS
+
+
+@mcp.tool()
+async def get_latest_news(ctx: Context, limit: int = 10) -> dict:
+    """Get the most recent crypto news articles, newest first.
+
+    Returns news with title text, source, link, related coins, AI rating, and tags.
+
+    Args:
+        limit: Maximum number of articles to return (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+    try:
+        result = await api.search_news(limit=limit, page=1)
+        data = result.get("data", [])[:limit]
+        return make_serializable({
+            "success": True, "data": data,
+            "count": len(data), "total": result.get("total", 0),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def search_news(keyword: str, ctx: Context, limit: int = 10) -> dict:
+    """Search crypto news by keyword in text content.
+
+    Args:
+        keyword: Search term (e.g. "bitcoin", "SEC", "ETF").
+        limit: Maximum results (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+    try:
+        result = await api.search_news(query=keyword, limit=limit, page=1)
+        data = result.get("data", [])[:limit]
+        return make_serializable({
+            "success": True, "keyword": keyword, "data": data,
+            "count": len(data), "total": result.get("total", 0),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def search_news_by_coin(coin: str, ctx: Context, limit: int = 10) -> dict:
+    """Search news related to a specific cryptocurrency coin/token.
+
+    Args:
+        coin: Coin symbol or name (e.g. "BTC", "ETH", "SOL", "TRUMP").
+        limit: Maximum results (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+    try:
+        result = await api.search_news(coins=[coin], limit=limit, page=1)
+        data = result.get("data", [])[:limit]
+        return make_serializable({
+            "success": True, "coin": coin, "data": data,
+            "count": len(data), "total": result.get("total", 0),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def get_news_by_source(engine_type: str, news_type: str, ctx: Context, limit: int = 10) -> dict:
+    """Get news articles from a specific source.
+
+    Use get_news_sources first to see available engine types and news type codes.
+
+    Args:
+        engine_type: The engine type (e.g. "news", "listing", "onchain", "meme", "market").
+        news_type: The news source code (e.g. "Bloomberg", "Reuters", "Coindesk").
+        limit: Maximum results (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+    try:
+        result = await api.search_news(engine_types={engine_type: [news_type]}, limit=limit, page=1)
+        data = result.get("data", [])[:limit]
+        return make_serializable({
+            "success": True, "engine_type": engine_type, "news_type": news_type, "data": data,
+            "count": len(data), "total": result.get("total", 0),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def get_news_by_engine(engine_type: str, ctx: Context, limit: int = 10) -> dict:
+    """Get news articles filtered by engine type.
+
+    Engine types: "news", "listing", "onchain", "meme", "market".
+
+    Args:
+        engine_type: The engine type code.
+        limit: Maximum results (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+    try:
+        result = await api.search_news(engine_types={engine_type: []}, limit=limit, page=1)
+        data = result.get("data", [])[:limit]
+        return make_serializable({
+            "success": True, "engine_type": engine_type, "data": data,
+            "count": len(data), "total": result.get("total", 0),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def search_news_advanced(
+    ctx: Context,
+    coins: str = "",
+    keyword: str = "",
+    engine_types: str = "",
+    has_coin: bool = False,
+    limit: int = 10,
+) -> dict:
+    """Advanced news search with multiple filters.
+
+    Args:
+        coins: Comma-separated coin symbols (e.g. "BTC,ETH").
+        keyword: Optional search keyword.
+        engine_types: Engine type filter in format "type1:cat1,cat2;type2:cat3" (e.g. "news:Bloomberg,Reuters;listing:").
+        has_coin: If true, only return news that have associated coins.
+        limit: Maximum results (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+
+    coin_list = [c.strip() for c in coins.split(",") if c.strip()] if coins else None
+
+    # 解析 engine_types 字符串为 dict
+    engine_types_dict = None
+    if engine_types:
+        engine_types_dict = {}
+        for part in engine_types.split(";"):
+            if ":" in part:
+                engine, cats = part.split(":", 1)
+                engine = engine.strip()
+                cat_list = [c.strip() for c in cats.split(",") if c.strip()]
+                engine_types_dict[engine] = cat_list
+
+    try:
+        result = await api.search_news(
+            coins=coin_list, query=keyword or None,
+            engine_types=engine_types_dict, has_coin=has_coin,
+            limit=limit, page=1,
+        )
+        data = result.get("data", [])[:limit]
+        return make_serializable({
+            "success": True, "data": data,
+            "count": len(data), "total": result.get("total", 0),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def get_high_score_news(ctx: Context, min_score: int = 70, limit: int = 10) -> dict:
+    """Get highly-rated news articles (by AI score), sorted by score descending.
+
+    Args:
+        min_score: Minimum score threshold (default 70).
+        limit: Maximum results to return (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+    try:
+        fetch_limit = min(limit * 3, MAX_ROWS)
+        result = await api.search_news(limit=fetch_limit, page=1)
+        raw = result.get("data", [])
+
+        filtered = [it for it in raw
+                     if (it.get("aiRating") or {}).get("score", 0) >= min_score]
+        filtered.sort(
+            key=lambda x: (x.get("aiRating") or {}).get("score", 0),
+            reverse=True,
+        )
+        data = filtered[:limit]
+        return make_serializable({
+            "success": True, "min_score": min_score,
+            "data": data, "count": len(data),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+
+
+@mcp.tool()
+async def get_news_by_signal(signal: str, ctx: Context, limit: int = 10) -> dict:
+    """Get news filtered by trading signal type.
+
+    Args:
+        signal: The signal type: "long" (bullish), "short" (bearish), or "neutral".
+        limit: Maximum results (default 10, max 100).
+    """
+    api = ctx.request_context.lifespan_context.api
+    limit = clamp_limit(limit)
+    try:
+        fetch_limit = min(limit * 3, MAX_ROWS)
+        result = await api.search_news(limit=fetch_limit, page=1)
+        raw = result.get("data", [])
+
+        filtered = [it for it in raw
+                     if (it.get("aiRating") or {}).get("signal") == signal
+                     and (it.get("aiRating") or {}).get("status") == "done"]
+        data = filtered[:limit]
+        return make_serializable({
+            "success": True, "signal": signal,
+            "data": data, "count": len(data),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}

src/opennews_mcp/tools/realtime.py

@@ -0,0 +1,68 @@
+"""Real-time news tools — WebSocket subscription for live news updates."""
+
+from mcp.server.fastmcp import Context
+
+from opennews_mcp.app import mcp
+from opennews_mcp.config import make_serializable
+
+
+@mcp.tool()
+async def subscribe_latest_news(
+    ctx: Context,
+    wait_seconds: int = 10,
+    max_items: int = 5,
+    coins: str = "",
+    engine_types: str = "",
+    has_coin: bool = False,
+) -> dict:
+    """Subscribe to real-time news updates via WebSocket.
+
+    Connects to the WebSocket feed, subscribes to news with optional filters,
+    and collects incoming messages for the specified duration.
+
+    Args:
+        wait_seconds: How long to listen for news (default 10, max 30 seconds).
+        max_items: Maximum news items to collect (default 5, max 20).
+        coins: Comma-separated coin symbols to filter (e.g. "BTC,ETH").
+        engine_types: Engine type filter in format "type1:cat1,cat2;type2:cat3".
+        has_coin: If true, only receive news that have associated coins.
+    """
+    ws = ctx.request_context.lifespan_context.ws
+    wait_seconds = min(max(1, wait_seconds), 30)
+    max_items = min(max(1, max_items), 20)
+
+    # 解析过滤器参数
+    coin_list = [c.strip() for c in coins.split(",") if c.strip()] if coins else None
+    engine_types_dict = None
+    if engine_types:
+        engine_types_dict = {}
+        for part in engine_types.split(";"):
+            if ":" in part:
+                engine, cats = part.split(":", 1)
+                engine = engine.strip()
+                cat_list = [c.strip() for c in cats.split(",") if c.strip()]
+                engine_types_dict[engine] = cat_list
+
+    try:
+        sub_result = await ws.subscribe_latest(
+            engine_types=engine_types_dict,
+            coins=coin_list,
+            has_coin=has_coin,
+        )
+
+        items = []
+        for _ in range(max_items):
+            msg = await ws.receive_news(timeout=float(wait_seconds))
+            if msg is None:
+                break
+            items.append(msg)
+
+        return make_serializable({
+            "success": True,
+            "data": items,
+            "count": len(items),
+        })
+    except Exception as e:
+        return {"success": False, "error": str(e) or repr(e)}
+    finally:
+        await ws.close()