mcp-miniflux/server.py

"""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)