graylog_mcp.py

"""Graylog MCP Server - Expose Graylog log management via MCP."""

import os
from typing import Annotated

import httpx
from mcp.server.fastmcp import FastMCP

# Configuration from environment
GRAYLOG_URL = os.environ.get("GRAYLOG_URL", "").rstrip("/")
GRAYLOG_TOKEN = os.environ.get("GRAYLOG_TOKEN", "")
GRAYLOG_TIMEOUT = int(os.environ.get("GRAYLOG_TIMEOUT", "30"))

# Initialize MCP server
mcp = FastMCP("graylog")


class GraylogClient:
    """Async HTTP client for Graylog API."""

    def __init__(self):
        self._client: httpx.AsyncClient | None = None

    async def get_client(self) -> httpx.AsyncClient:
        """Get or create the HTTP client."""
        if self._client is None or self._client.is_closed:
            if not GRAYLOG_URL:
                raise ValueError("GRAYLOG_URL environment variable is required")
            if not GRAYLOG_TOKEN:
                raise ValueError("GRAYLOG_TOKEN environment variable is required")

            self._client = httpx.AsyncClient(
                base_url=GRAYLOG_URL,
                auth=(GRAYLOG_TOKEN, "token"),
                headers={
                    "Accept": "application/json",
                    "X-Requested-By": "graylog-mcp",
                },
                timeout=GRAYLOG_TIMEOUT,
            )
        return self._client

    async def get(self, endpoint: str, params: dict | None = None) -> dict:
        """Make a GET request to Graylog API."""
        client = await self.get_client()
        try:
            response = await client.get(endpoint, params=params)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 401:
                raise ValueError("Authentication failed. Check GRAYLOG_TOKEN.")
            elif e.response.status_code == 404:
                raise ValueError(f"Resource not found: {endpoint}")
            else:
                raise ValueError(f"Graylog API error: {e.response.status_code}")
        except httpx.ConnectError:
            raise ValueError(f"Cannot connect to Graylog at {GRAYLOG_URL}")
        except httpx.TimeoutException:
            raise ValueError(f"Request timed out after {GRAYLOG_TIMEOUT}s")


# Global client instance
graylog = GraylogClient()


def parse_timerange(timerange: str) -> int:
    """Parse a human-readable timerange into seconds.

    Examples: 'last 5 minutes', 'last 1 hour', 'last 24 hours', 'last 7 days'
    """
    timerange = timerange.lower().strip()
    if timerange.startswith("last "):
        timerange = timerange[5:]

    parts = timerange.split()
    if len(parts) >= 2:
        try:
            value = int(parts[0])
            unit = parts[1].rstrip('s')  # Remove trailing 's'
            multipliers = {
                'second': 1,
                'minute': 60,
                'hour': 3600,
                'day': 86400,
                'week': 604800,
            }
            if unit in multipliers:
                return value * multipliers[unit]
        except ValueError:
            pass

    # Default to 15 minutes
    return 900


@mcp.tool()
async def search_logs(
    query: Annotated[str, "Graylog search query (e.g., 'level:ERROR AND source:nginx')"],
    timerange: Annotated[str, "Time range like 'last 5 minutes', 'last 1 hour', 'last 24 hours'"] = "last 15 minutes",
    limit: Annotated[int, "Maximum results to return (1-1000)"] = 100,
    fields: Annotated[str | None, "Comma-separated fields to return (e.g., 'message,source,timestamp')"] = None,
    stream_id: Annotated[str | None, "Filter to specific stream ID"] = None,
) -> dict:
    """Search Graylog logs with relative time ranges."""
    range_seconds = parse_timerange(timerange)

    params = {
        "query": query,
        "range": range_seconds,
        "limit": min(max(1, limit), 1000),
        "sort": "timestamp:desc",
        "fields": fields if fields else "message,source,timestamp",
    }
    if stream_id:
        params["filter"] = f"streams:{stream_id}"

    response = await graylog.get("/search/universal/relative", params)

    messages = response.get("messages", [])
    return {
        "total_results": response.get("total_results", 0),
        "query": query,
        "timerange": timerange,
        "messages": [
            {
                "timestamp": msg.get("message", {}).get("timestamp"),
                "source": msg.get("message", {}).get("source"),
                "message": msg.get("message", {}).get("message"),
                "level": msg.get("message", {}).get("level"),
            }
            for msg in messages
        ],
    }


def normalize_timestamp(ts: str) -> str:
    """Convert various timestamp formats to Graylog's expected format.

    Graylog expects: 'YYYY-MM-DD HH:MM:SS'
    Accepts: ISO 8601 ('2024-01-15T10:00:00Z') or already formatted.
    """
    # Replace T separator with space, remove Z suffix and timezone
    ts = ts.replace("T", " ").rstrip("Z")
    # Remove milliseconds if present
    if "." in ts:
        ts = ts.split(".")[0]
    # Remove timezone offset if present (+00:00, -05:00, etc.)
    if "+" in ts:
        ts = ts.split("+")[0]
    elif ts.count("-") > 2:  # Has timezone like -05:00
        parts = ts.rsplit("-", 1)
        if ":" in parts[-1]:
            ts = parts[0]
    return ts.strip()


@mcp.tool()
async def search_logs_absolute(
    query: Annotated[str, "Graylog search query"],
    from_time: Annotated[str, "Start time (e.g., '2024-01-15 10:00:00' or '2024-01-15T10:00:00Z')"],
    to_time: Annotated[str, "End time (e.g., '2024-01-15 11:00:00' or '2024-01-15T11:00:00Z')"],
    limit: Annotated[int, "Maximum results to return (1-1000)"] = 100,
    fields: Annotated[str | None, "Comma-separated fields to return (e.g., 'message,source,timestamp')"] = None,
    stream_id: Annotated[str | None, "Filter to specific stream ID"] = None,
) -> dict:
    """Search Graylog logs with absolute time range."""
    params = {
        "query": query,
        "from": normalize_timestamp(from_time),
        "to": normalize_timestamp(to_time),
        "limit": min(max(1, limit), 1000),
        "sort": "timestamp:desc",
        "fields": fields if fields else "message,source,timestamp",
    }
    if stream_id:
        params["filter"] = f"streams:{stream_id}"

    response = await graylog.get("/search/universal/absolute", params)

    messages = response.get("messages", [])
    return {
        "total_results": response.get("total_results", 0),
        "query": query,
        "from": params["from"],
        "to": params["to"],
        "messages": [
            {
                "timestamp": msg.get("message", {}).get("timestamp"),
                "source": msg.get("message", {}).get("source"),
                "message": msg.get("message", {}).get("message"),
                "level": msg.get("message", {}).get("level"),
            }
            for msg in messages
        ],
    }


@mcp.tool()
async def list_streams() -> list[dict]:
    """List all available Graylog streams."""
    response = await graylog.get("/streams")
    streams = response.get("streams", [])
    return [
        {
            "id": stream.get("id"),
            "title": stream.get("title"),
            "description": stream.get("description"),
            "disabled": stream.get("disabled", False),
        }
        for stream in streams
    ]


@mcp.tool()
async def get_stream_details(
    stream_id: Annotated[str, "Stream ID to retrieve details for"],
) -> dict:
    """Get detailed information about a specific Graylog stream."""
    return await graylog.get(f"/streams/{stream_id}")


@mcp.tool()
async def list_alerts() -> list[dict]:
    """List all alert conditions configured in Graylog."""
    response = await graylog.get("/alerts/conditions")
    conditions = response.get("conditions", [])
    return [
        {
            "id": cond.get("id"),
            "title": cond.get("title"),
            "type": cond.get("type"),
            "in_grace": cond.get("in_grace", False),
        }
        for cond in conditions
    ]


@mcp.tool()
async def get_system_info() -> dict:
    """Get Graylog cluster and system information."""
    return await graylog.get("/system/cluster")


@mcp.tool()
async def list_dashboards() -> list[dict]:
    """List all Graylog dashboards."""
    response = await graylog.get("/dashboards")
    dashboards = response.get("dashboards", [])
    return [
        {
            "id": dash.get("id"),
            "title": dash.get("title"),
            "description": dash.get("description"),
        }
        for dash in dashboards
    ]


@mcp.tool()
async def get_dashboard_details(
    dashboard_id: Annotated[str, "Dashboard ID to retrieve"],
) -> dict:
    """Get detailed information about a specific dashboard including widgets."""
    return await graylog.get(f"/dashboards/{dashboard_id}")


if __name__ == "__main__":
    mcp.run(transport="stdio")