Add Miniflux FastMCP server

server.py

@@ -0,0 +1,158 @@
+"""Miniflux MCP server — read, triage, and manage RSS feeds via the Miniflux API."""
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+from fastmcp import FastMCP
+
+BASE_URL = os.environ.get("MINIFLUX_URL", "http://miniflux:8080")
+API_KEY  = os.environ.get("MINIFLUX_API_KEY", "")
+
+mcp = FastMCP(
+    "miniflux",
+    instructions=(
+        "Tools for the Miniflux RSS reader. Use get_unread_entries to surface new articles, "
+        "mark_entries_read to triage them, list_feeds to see subscriptions, and add_feed to subscribe. "
+        "Entries have id, title, url, feed_title, published_at, content (HTML), and reading_time fields."
+    ),
+)
+
+
+def _client() -> httpx.AsyncClient:
+    return httpx.AsyncClient(
+        base_url=BASE_URL,
+        headers={"X-Auth-Token": API_KEY, "Accept": "application/json"},
+        timeout=30.0,
+    )
+
+
+async def _get(path: str, params: dict | None = None) -> Any:
+    async with _client() as c:
+        r = await c.get(path, params={k: v for k, v in (params or {}).items() if v is not None})
+        r.raise_for_status()
+        return r.json()
+
+
+async def _post(path: str, body: dict | None = None) -> Any:
+    async with _client() as c:
+        r = await c.post(path, json=body or {})
+        r.raise_for_status()
+        return r.json() if r.content else {"ok": True}
+
+
+async def _put(path: str, body: dict) -> Any:
+    async with _client() as c:
+        r = await c.put(path, json=body)
+        r.raise_for_status()
+        return r.json() if r.content else {"ok": True}
+
+
+def _fmt_entry(e: dict) -> dict:
+    return {
+        "id": e["id"],
+        "title": e.get("title", ""),
+        "url": e.get("url", ""),
+        "feed": e.get("feed", {}).get("title", ""),
+        "published_at": e.get("published_at", ""),
+        "reading_time": e.get("reading_time", 0),
+        "status": e.get("status", ""),
+    }
+
+
+@mcp.tool()
+async def get_unread_entries(limit: int = 20, feed_id: int | None = None) -> list[dict]:
+    """Get unread RSS entries. Returns id, title, url, feed, published_at, reading_time."""
+    params: dict = {"status": "unread", "limit": limit, "order": "published_at", "direction": "desc"}
+    if feed_id:
+        params["feed_id"] = feed_id
+    data = await _get("/v1/entries", params)
+    return [_fmt_entry(e) for e in data.get("entries", [])]
+
+
+@mcp.tool()
+async def get_entry_content(entry_id: int) -> dict:
+    """Get the full content of a specific entry by id."""
+    e = await _get(f"/v1/entries/{entry_id}")
+    return {
+        "id": e["id"],
+        "title": e.get("title", ""),
+        "url": e.get("url", ""),
+        "feed": e.get("feed", {}).get("title", ""),
+        "published_at": e.get("published_at", ""),
+        "content": e.get("content", ""),
+        "author": e.get("author", ""),
+    }
+
+
+@mcp.tool()
+async def mark_entries_read(entry_ids: list[int]) -> dict:
+    """Mark one or more entries as read by their ids."""
+    await _put("/v1/entries", {"entry_ids": entry_ids, "status": "read"})
+    return {"marked_read": len(entry_ids)}
+
+
+@mcp.tool()
+async def mark_entries_unread(entry_ids: list[int]) -> dict:
+    """Mark one or more entries as unread by their ids."""
+    await _put("/v1/entries", {"entry_ids": entry_ids, "status": "unread"})
+    return {"marked_unread": len(entry_ids)}
+
+
+@mcp.tool()
+async def list_feeds() -> list[dict]:
+    """List all subscribed RSS feeds with id, title, site_url, unread count, and last check."""
+    feeds = await _get("/v1/feeds")
+    return [
+        {
+            "id": f["id"],
+            "title": f.get("title", ""),
+            "site_url": f.get("site_url", ""),
+            "feed_url": f.get("feed_url", ""),
+            "unread_count": f.get("unread_count", 0),
+            "last_checked": f.get("checked_at", ""),
+            "category": f.get("category", {}).get("title", ""),
+        }
+        for f in (feeds or [])
+    ]
+
+
+@mcp.tool()
+async def add_feed(feed_url: str, category_id: int | None = None) -> dict:
+    """Subscribe to a new RSS/Atom feed by URL. Optionally assign a category_id."""
+    body: dict = {"feed_url": feed_url}
+    if category_id:
+        body["category_id"] = category_id
+    result = await _post("/v1/feeds", body)
+    return result
+
+
+@mcp.tool()
+async def refresh_feed(feed_id: int) -> dict:
+    """Trigger an immediate refresh of a feed."""
+    async with _client() as c:
+        r = await c.put(f"/v1/feeds/{feed_id}/refresh")
+        r.raise_for_status()
+    return {"refreshed": feed_id}
+
+
+@mcp.tool()
+async def search_entries(query: str, limit: int = 20, status: str = "all") -> list[dict]:
+    """Full-text search across entry titles and content. status: unread | read | all."""
+    params: dict = {"search": query, "limit": limit}
+    if status != "all":
+        params["status"] = status
+    data = await _get("/v1/entries", params)
+    return [_fmt_entry(e) for e in data.get("entries", [])]
+
+
+@mcp.tool()
+async def list_categories() -> list[dict]:
+    """List all feed categories."""
+    cats = await _get("/v1/categories")
+    return [{"id": c["id"], "title": c.get("title", "")} for c in (cats or [])]
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)