pastebin-mcp/mcp_server.py

"""
pastebin-mcp — MCP server for the homelab pastebin service.

Wraps the pastebin REST API so agents can create, retrieve, list, and
delete pastes programmatically.

Configure via environment variables:
  PASTEBIN_URL     — Pastebin base URL (default http://192.168.0.242:3002)
  PASTEBIN_API_KEY — API key for privileged operations (list, delete)

Transport:
  PASTEBIN_MCP_TRANSPORT — stdio (default), sse, or streamable-http
  FASTMCP_HOST           — bind host for HTTP transports (default 127.0.0.1)
  FASTMCP_PORT           — bind port for HTTP transports (default 8000)
  FASTMCP_STREAMABLE_HTTP_PATH — path for streamable-http (default /mcp)
"""

import os
from typing import Optional

import httpx
from mcp.server.fastmcp import FastMCP

PASTEBIN_URL = os.environ.get("PASTEBIN_URL", "http://192.168.0.242:3002").rstrip("/")
PASTEBIN_API_KEY = os.environ.get("PASTEBIN_API_KEY", "")

MCP_TRANSPORT = os.getenv("PASTEBIN_MCP_TRANSPORT", "stdio").strip().lower() or "stdio"
MCP_HOST      = os.getenv("FASTMCP_HOST", "127.0.0.1").strip() or "127.0.0.1"
MCP_PORT      = int(os.getenv("FASTMCP_PORT", "8000"))
MCP_PATH      = os.getenv("FASTMCP_STREAMABLE_HTTP_PATH", "/mcp").strip() or "/mcp"

mcp = FastMCP(
    "pastebin-mcp",
    host=MCP_HOST,
    port=MCP_PORT,
    streamable_http_path=MCP_PATH,
    json_response=True,
    stateless_http=True,
)

# ---------------------------------------------------------------------------
# HTTP helpers
# ---------------------------------------------------------------------------

def _client() -> httpx.Client:
    """Return an httpx client pointed at the pastebin base URL."""
    return httpx.Client(base_url=PASTEBIN_URL, timeout=10.0)


def _authed_client() -> httpx.Client:
    """Return an httpx client with the API key header set for privileged ops."""
    return httpx.Client(
        base_url=PASTEBIN_URL,
        headers={"X-API-Key": PASTEBIN_API_KEY},
        timeout=10.0,
    )


def _check(r: httpx.Response) -> dict | list | str:
    """Raise on non-2xx responses; return parsed JSON or raw text."""
    if not r.is_success:
        raise RuntimeError(f"Pastebin API error {r.status_code}: {r.text}")
    content_type = r.headers.get("content-type", "")
    if "application/json" in content_type:
        return r.json()
    return r.text


# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------

@mcp.tool()
def create_paste(
    content: str,
    language: str = "plaintext",
    title: str = "",
    expires_in_hours: Optional[int] = None,
    private: bool = False,
) -> dict:
    """
    Create a new paste and return its URL and ID.

    No authentication is required to create a paste. The paste is immediately
    accessible at the returned URL.

    Args:
        content:          The text content of the paste. May be any UTF-8 string.
        language:         Syntax-highlighting language hint, e.g. ``"python"``,
                          ``"javascript"``, ``"yaml"``, ``"plaintext"`` (default).
                          The pastebin service uses this for display only.
        title:            Optional human-readable title for the paste. Empty string
                          means no title (default).
        expires_in_hours: Optional number of hours until the paste expires and is
                          automatically deleted. ``None`` (default) means the paste
                          never expires.
        private:          If ``True``, the paste will not appear in public listings.
                          Defaults to ``False``.

    Returns:
        Dict with keys:
          - ``id``      — unique paste identifier (used in other tool calls)
          - ``url``     — full public URL to view the paste
          - ``raw_url`` — direct URL to the raw paste content
    """
    payload: dict = {"content": content, "language": language, "private": private}
    if title:
        payload["title"] = title
    if expires_in_hours is not None:
        payload["expires_in_hours"] = expires_in_hours

    with _client() as c:
        result: dict = _check(c.post("/api/paste", json=payload))  # type: ignore[assignment]

    paste_id = result.get("id", "")
    return {
        "id":      paste_id,
        "url":     f"{PASTEBIN_URL}/paste/{paste_id}" if paste_id else "",
        "raw_url": f"{PASTEBIN_URL}/raw/{paste_id}" if paste_id else "",
        **result,
    }


@mcp.tool()
def get_paste(paste_id: str) -> dict:
    """
    Retrieve the full metadata and content for a single paste.

    No authentication is required to read a paste.

    Args:
        paste_id: The unique identifier of the paste, as returned by
                  ``create_paste`` or ``list_pastes``.

    Returns:
        Full paste object with keys including:
          - ``id``         — paste identifier
          - ``content``    — full text content
          - ``language``   — syntax language hint
          - ``title``      — paste title (may be empty)
          - ``private``    — whether the paste is private
          - ``created_at`` — ISO-8601 creation timestamp
          - ``expires_at`` — ISO-8601 expiry timestamp, or ``null`` if never
    """
    with _client() as c:
        return _check(c.get(f"/api/paste/{paste_id}"))  # type: ignore[return-value]


@mcp.tool()
def get_raw(paste_id: str) -> str:
    """
    Retrieve the raw text content of a paste as a plain string.

    Useful when you only need the content itself without metadata overhead.
    No authentication is required.

    Args:
        paste_id: The unique identifier of the paste, as returned by
                  ``create_paste`` or ``list_pastes``.

    Returns:
        The raw text content of the paste as a plain string.
    """
    with _client() as c:
        return _check(c.get(f"/raw/{paste_id}"))  # type: ignore[return-value]


@mcp.tool()
def list_pastes(limit: int = 50, offset: int = 0) -> list:
    """
    List recent pastes from the pastebin service.

    Requires the ``PASTEBIN_API_KEY`` environment variable to be set.
    Returns pastes in reverse-chronological order (newest first).

    Args:
        limit:  Maximum number of pastes to return (default 50). The server
                may cap this at a lower value.
        offset: Zero-based offset for pagination (default 0). Use in
                combination with ``limit`` to page through results.

    Returns:
        List of paste summary objects, each containing:
          - ``id``         — paste identifier
          - ``title``      — paste title (may be empty)
          - ``language``   — syntax language hint
          - ``private``    — whether the paste is private
          - ``created_at`` — ISO-8601 creation timestamp
          - ``expires_at`` — ISO-8601 expiry timestamp, or ``null``
        Note: the ``content`` field is typically omitted from listing responses
        to reduce payload size. Use ``get_paste`` to fetch full content.
    """
    with _authed_client() as c:
        return _check(c.get("/api/pastes", params={"limit": limit, "offset": offset}))  # type: ignore[return-value]


@mcp.tool()
def delete_paste(paste_id: str) -> dict:
    """
    Permanently delete a paste.

    Requires the ``PASTEBIN_API_KEY`` environment variable to be set.
    This operation is irreversible.

    Args:
        paste_id: The unique identifier of the paste to delete, as returned
                  by ``create_paste`` or ``list_pastes``.

    Returns:
        Dict with:
          - ``deleted``  — ``True`` if the paste was successfully deleted
          - ``paste_id`` — the ID of the deleted paste
        Raises ``RuntimeError`` if the paste does not exist or the API key
        is missing or invalid.
    """
    with _authed_client() as c:
        r = c.delete(f"/api/paste/{paste_id}")
    if not r.is_success:
        raise RuntimeError(f"Pastebin API error {r.status_code}: {r.text}")
    return {"deleted": True, "paste_id": paste_id}


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

MCP_API_KEY = os.environ.get("MCP_API_KEY", "").strip()


class _BearerAuthMiddleware:
    """Pure-ASGI middleware rejecting HTTP requests without a valid bearer token.

    Enforced only when MCP_API_KEY is set; stdio transport is unaffected.
    """

    def __init__(self, app, token: str) -> None:
        self.app = app
        self._expected = f"Bearer {token}".encode()

    async def __call__(self, scope, receive, send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        headers = dict(scope.get("headers") or [])
        if headers.get(b"authorization") != self._expected:
            from starlette.responses import JSONResponse

            await JSONResponse(
                {"jsonrpc": "2.0",
                 "error": {"code": -32001, "message": "Unauthorized"},
                 "id": None},
                status_code=401,
                headers={"WWW-Authenticate": "Bearer"},
            )(scope, receive, send)
            return
        await self.app(scope, receive, send)


def main() -> None:
    if MCP_TRANSPORT not in {"stdio", "sse", "streamable-http"}:
        raise ValueError("PASTEBIN_MCP_TRANSPORT must be stdio, sse, or streamable-http.")
    if MCP_TRANSPORT == "streamable-http" and MCP_API_KEY:
        import uvicorn

        app = mcp.streamable_http_app()
        app.add_middleware(_BearerAuthMiddleware, token=MCP_API_KEY)
        uvicorn.run(app, host=MCP_HOST, port=MCP_PORT)
    else:
        mcp.run(transport=MCP_TRANSPORT)


if __name__ == "__main__":
    main()