用自定义 MCP 接入企业日志系统——从命令行到一键定位问题

Python >= 3.10
mcp[cli] == 1.27.2（2026-05-29 发布，截至本讲写作时的最新版）
httpx == 0.28.x（异步 HTTP 客户端）

pip install "mcp[cli]" httpx

# loki_mcp/server.py
import os
import base64
from datetime import datetime, timedelta, timezone

import httpx
from mcp.server.fastmcp import FastMCP

# --- 初始化 FastMCP server ---
mcp = FastMCP("loki-log-server")

# --- 从环境变量读取认证配置 ---
LOKI_URL = os.environ.get("LOKI_URL", "http://localhost:3100")
LOKI_USERNAME = os.environ.get("LOKI_USERNAME", "")
LOKI_PASSWORD = os.environ.get("LOKI_PASSWORD", "")
LOKI_BEARER_TOKEN = os.environ.get("LOKI_BEARER_TOKEN", "")
LOKI_ORG_ID = os.environ.get("LOKI_ORG_ID", "")  # Grafana 多租户头

# --- 查询安全限制 ---
MAX_LIMIT = int(os.environ.get("LOKI_MAX_LIMIT", "1000"))  # 单次最多返回条数
MAX_HOURS = int(os.environ.get("LOKI_MAX_HOURS", "24"))    # 最多回溯小时数


def _build_headers() -> dict[str, str]:
    """根据环境变量构建认证 headers，优先级：Bearer > Basic > 无认证"""
    headers = {"Content-Type": "application/json"}
    
    if LOKI_BEARER_TOKEN:
        headers["Authorization"] = f"Bearer {LOKI_BEARER_TOKEN}"
    elif LOKI_USERNAME and LOKI_PASSWORD:
        credentials = base64.b64encode(
            f"{LOKI_USERNAME}:{LOKI_PASSWORD}".encode()
        ).decode()
        headers["Authorization"] = f"Basic {credentials}"
    
    if LOKI_ORG_ID:
        headers["X-Scope-OrgID"] = LOKI_ORG_ID
    
    return headers


def _validate_time_range(start_hours: float, end_hours: float) -> tuple[str, str]:
    """
    把"N 小时前"转成 Loki 需要的 nanosecond 时间戳字符串。
    同时做安全校验：时间范围不超过 MAX_HOURS。
    """
    if start_hours < 0or end_hours < 0:
        raise ValueError("时间范围不能是负数")
    if start_hours > MAX_HOURS:
        raise ValueError(f"最多只能查 {MAX_HOURS} 小时内的日志，当前请求 {start_hours} 小时")
    
    now = datetime.now(timezone.utc)
    start_dt = now - timedelta(hours=start_hours)
    end_dt = now - timedelta(hours=end_hours)
    
    # Loki API 使用 nanosecond 时间戳
    start_ns = str(int(start_dt.timestamp() * 1e9))
    end_ns = str(int(end_dt.timestamp() * 1e9))
    
    return start_ns, end_ns

@mcp.tool()
asyncdef list_labels() -> dict:
    """
    列出 Loki 中所有可用的 label 名称。
    在构造查询之前，先用这个工具了解当前日志系统的结构。
    
    返回示例：{"labels": ["app", "env", "namespace", "pod", "level"]}
    """
    asyncwith httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(
            f"{LOKI_URL}/loki/api/v1/labels",
            headers=_build_headers(),
            verify=False# 内网自签名证书常见，生产环境替换为 CA 路径
        )
        resp.raise_for_status()
        data = resp.json()
        return {"labels": data.get("data", [])}


@mcp.tool()
asyncdef list_label_values(label_name: str) -> dict:
    """
    列出指定 label 的所有可能值。
    例如：list_label_values("app") 返回 ["order-service", "payment-service", "gateway"]
    
    Args:
        label_name: label 名称，必须是 list_labels 返回的 label 之一
    """
    ifnot label_name ornot label_name.replace("-", "").replace("_", "").isalnum():
        raise ValueError(f"无效的 label 名称: {label_name!r}")
    
    asyncwith httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(
            f"{LOKI_URL}/loki/api/v1/label/{label_name}/values",
            headers=_build_headers(),
            verify=False
        )
        resp.raise_for_status()
        data = resp.json()
        return {
            "label": label_name,
            "values": data.get("data", [])
        }


@mcp.tool()
asyncdef list_streams(
    match: str = "",
    start_hours: float = 1.0
) -> dict:
    """
    列出当前活跃的日志流（流 = 一组 label 的组合）。
    
    Args:
        match: LogQL stream selector，例如 '{app="order-service"}'，
               空字符串返回所有流（数量可能很大，建议加过滤）
        start_hours: 往前看多少小时，默认 1 小时
    """
    start_ns, end_ns = _validate_time_range(start_hours, 0)
    
    params: dict = {"start": start_ns, "end": end_ns, "limit": "100"}
    if match:
        params["match[]"] = match
    
    asyncwith httpx.AsyncClient(timeout=15.0) as client:
        resp = await client.get(
            f"{LOKI_URL}/loki/api/v1/series",
            headers=_build_headers(),
            params=params,
            verify=False
        )
        resp.raise_for_status()
        data = resp.json()
        return {
            "streams": data.get("data", []),
            "count": len(data.get("data", []))
        }

@mcp.tool()
asyncdef query_logs(
    logql: str,
    start_hours: float = 1.0,
    end_hours: float = 0.0,
    limit: int = 100,
    direction: str = "backward"
) -> dict:
    """
    按时间范围查询日志流。这是最常用的查询工具。
    
    Args:
        logql: LogQL 查询语句，例如：
               '{app="order-service"} |= "ERROR"'
               '{namespace="production", level="error"} | json | duration > 1s'
        start_hours: 查询起点（N 小时前），默认 1 小时前
        end_hours: 查询终点（N 小时前），默认 0（即现在）
        limit: 最多返回条数，默认 100，最大 {MAX_LIMIT}
        direction: 排序方向，"backward"（最新的在前）或 "forward"
    """
    if direction notin ("backward", "forward"):
        raise ValueError("direction 必须是 'backward' 或 'forward'")
    
    limit = min(limit, MAX_LIMIT)
    start_ns, end_ns = _validate_time_range(start_hours, end_hours)
    
    params = {
        "query": logql,
        "start": start_ns,
        "end": end_ns,
        "limit": str(limit),
        "direction": direction
    }
    
    asyncwith httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.get(
            f"{LOKI_URL}/loki/api/v1/query_range",
            headers=_build_headers(),
            params=params,
            verify=False
        )
        resp.raise_for_status()
        data = resp.json()
    
    # 把 Loki 返回的格式整理成更易读的结构
    result_streams = data.get("data", {}).get("result", [])
    logs = []
    for stream in result_streams:
        labels = stream.get("stream", {})
        for ts_ns, log_line in stream.get("values", []):
            # 把 nanosecond 时间戳转成可读时间
            ts_sec = int(ts_ns) / 1e9
            readable_ts = datetime.fromtimestamp(
                ts_sec, tz=timezone.utc
            ).strftime("%Y-%m-%d %H:%M:%S.%f UTC")
            logs.append({
                "timestamp": readable_ts,
                "labels": labels,
                "line": log_line
            })
    
    return {
        "query": logql,
        "total": len(logs),
        "logs": logs
    }


@mcp.tool()
asyncdef instant_query(logql: str, time_hours_ago: float = 0.0) -> dict:
    """
    在某个特定时间点执行 LogQL 表达式（通常用于 metric 类查询）。
    
    Args:
        logql: LogQL metric 表达式，例如：
               'count_over_time({app="order-service"}[5m])'
               'rate({app="payment-service", level="error"}[1m])'
        time_hours_ago: 在多少小时前的时间点执行，0 表示当前时间
    """
    if time_hours_ago > MAX_HOURS:
        raise ValueError(f"时间点不能超过 {MAX_HOURS} 小时前")
    
    now = datetime.now(timezone.utc)
    query_time = now - timedelta(hours=time_hours_ago)
    query_time_ns = str(int(query_time.timestamp() * 1e9))
    
    asyncwith httpx.AsyncClient(timeout=15.0) as client:
        resp = await client.get(
            f"{LOKI_URL}/loki/api/v1/query",
            headers=_build_headers(),
            params={"query": logql, "time": query_time_ns},
            verify=False
        )
        resp.raise_for_status()
        data = resp.json()
    
    return {
        "query": logql,
        "result_type": data.get("data", {}).get("resultType"),
        "result": data.get("data", {}).get("result", [])
    }


# --- 入口 ---
if __name__ == "__main__":
    # stdio 模式，Claude Code 会把这个进程作为子进程启动
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "loki-logs": {
      "command": "python",
      "args": ["-m", "loki_mcp.server"],
      "env": {
        "LOKI_URL": "http://your-loki-server:3100",
        "LOKI_BEARER_TOKEN": "your-token-here",
        "LOKI_ORG_ID": "your-org-id",
        "LOKI_MAX_LIMIT": "500",
        "LOKI_MAX_HOURS": "48"
      }
    }
  }
}

/mcp                # 查看 MCP 工具列表，确认 loki-logs 加载成功

最近一小时订单服务有哪些 ERROR 日志？帮我分析一下根因。

{app="payment-service", error="true"}

{app="payment-service"} | json | level = "error"

@mcp.tool()
asyncdef get_log_schema() -> dict:
    """
    返回当前日志系统的结构说明和常用查询示例。
    在构造复杂查询之前，先调用这个工具了解日志格式。
    """
    return {
        "label_structure": {
            "app": "服务名，如 order-service, payment-service, gateway",
            "env": "环境，值为 production 或 staging",
            "namespace": "K8s namespace",
            "level": "注意：level 不是 label，需要通过 json 解析：| json | level = \"error\""
        },
        "common_queries": [
            '{app="order-service"} | json | level = "error"',
            '{app="payment-service", env="production"} |= "timeout" | json',
            'count_over_time({app="order-service"} | json | level = "error" [5m])'
        ],
        "gotchas": [
            "日志内容是 JSON 格式，level/trace_id 等字段需要先 | json 再用",
            "时间戳在 Loki 里精度到纳秒，API 返回的是纳秒字符串"
        ]
    }

# 在 query_logs 执行前，先做量级评估（伪代码）
count_query = f'count_over_time(({logql})[{time_range}])'
count_result = await _loki_instant_query(count_query)
if count_result > WARN_THRESHOLD:
    return {
        "warning": f"该查询预计返回 {count_result} 条日志，建议缩小时间范围或增加过滤条件",
        "suggestion": "缩小时间范围到 1 小时内，或增加 level='error' 过滤"
    }