234 lines
8.3 KiB
Python
234 lines
8.3 KiB
Python
"""
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main() -> None:
|
|
if MCP_TRANSPORT not in {"stdio", "sse", "streamable-http"}:
|
|
raise ValueError("PASTEBIN_MCP_TRANSPORT must be stdio, sse, or streamable-http.")
|
|
mcp.run(transport=MCP_TRANSPORT)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|