Kibana日志查询服务器

Kibana MCP Server

🔍 Kibana 日志查询 MCP Server - 提供自动认证、会话管理和多种日志查询工具

功能特性

✅ 自动认证管理

• 自动根据 KIBANA_USERNAME 和 KIBANA_PASSWORD 环境变量登录
• 自动获取并管理 sid cookie
• 会话过期自动续期
• 401 错误透明处理并重新登录

✅ 丰富的查询工具

• 日志搜索（关键词查询 + DSL 查询）
• 聚合统计（按服务、按时间）
• 错误日志快速搜索
• 原始 Elasticsearch DSL 查询

安装

# 1. 安装 Python 依赖
pip install -r requirements.txt

# 或使用 uv (推荐)
uv pip install -r requirements.txt

运行方式 (推荐使用 uvx)

你可以直接使用 uvx 从本地目录运行：

uvx --from /tmp/kibana-mcp-server kibana-mcp-server

配置 MCP

Claude Desktop / OpenCode 配置

{
  "mcpServers": {
    "kibana": {
      "command": "uvx",
      "args": [
        "--from",
        "/tmp/kibana-mcp-server",
        "kibana-mcp-server"
      ],
      "env": {
        "KIBANA_URL": "https://logs.example.com",
        "KIBANA_VERSION": "8.17.1",
        "KIBANA_USERNAME": "your_username",
        "KIBANA_PASSWORD": "your_password"
      }
    }
  }
}

1. kibana_search_logs

搜索日志（支持关键词搜索和完整 DSL 查询）

参数:

• query - 搜索关键词或完整 Elasticsearch DSL JSON
• time_range - 时间范围，默认 "now-1h"
• size - 返回数量，默认 20
• index_pattern - 索引模式，默认 "logstash-*"
• fields - 返回字段列表

示例 1: 关键词搜索

{
  "query": "user_id",
  "time_range": "now-30m",
  "size": 10
}

示例 2: DSL 查询

{
  "query": "{\"query\":{\"bool\":{\"must\":[{\"match\":{\"kubernetes.container_name\":\"api-server\"}}]}}}",
  "time_range": "now-1h",
  "size": 50
}

3. kibana_aggregate_logs

聚合统计日志

参数:

• aggregation_type - 聚合类型: "by_service" | "by_time" | "custom"
• time_range - 时间范围，默认 "now-1h"
• filter - 可选过滤条件（DSL JSON）
• custom_aggregation - 自定义聚合（仅 custom 类型）
• index_pattern - 索引模式

示例 1: 按服务统计

{
  "aggregation_type": "by_service",
  "time_range": "now-24h"
}

示例 2: 按时间统计

{
  "aggregation_type": "by_time",
  "time_range": "now-6h"
}

4. kibana_get_latest_logs

快速获取最新日志

参数:

• service - 服务名称（可选）
• size - 返回数量，默认 10
• index_pattern - 索引模式

示例:

{
  "service": "proxy-gateway",
  "size": 20
}

5. kibana_search_errors

搜索错误日志

参数:

• service - 服务名称（可选）
• severity - 严重程度: "error" | "exception" | "critical" | "all"
• time_range - 时间范围，默认 "now-1h"
• size - 返回数量，默认 20
• index_pattern - 索引模式

示例:

{
  "service": "api-server",
  "severity": "exception",
  "time_range": "now-2h",
  "size": 30
}

6. kibana_raw_query

执行原始 Elasticsearch DSL 查询（高级）

参数:

• path - Elasticsearch 路径，默认 "/logstash-*/_search"
• query - 完整的 Elasticsearch DSL JSON 字符串

示例:

{
  "path": "/logstash-*/_search",
  "query": "{\"query\":{\"match_all\":{}},\"size\":5}"
}

使用流程

1. 配置环境变量

在 MCP 配置文件中设置 KIBANA_USERNAME 和 KIBANA_PASSWORD。

2. 查询日志

直接调用工具即可，系统会自动处理登录：

使用工具: kibana_search_logs
参数:
  query: "error"
  time_range: "now-1h"
  size: 20

3. 统计分析

使用工具: kibana_aggregate_logs
参数:
  aggregation_type: "by_service"
  time_range: "now-24h"

技术细节

认证机制

1. 系统从环境变量加载 KIBANA_USERNAME 和 KIBANA_PASSWORD
2. 首次请求时自动登录 Kibana
3. 获取并保存 sid cookie（Iron 加密格式）
4. 后续请求自动携带 cookie
5. 遇到 401 错误时自动重新登录

会话管理

• 会话超时: 23 小时（避免 24 小时边界问题）
• 自动检测: 每次请求前检查会话有效性
• 透明续期: 用户无需关心会话状态

错误处理

• ✅ 401 Unauthorized → 自动重新登录
• ✅ 网络超时 → 30 秒超时配置
• ✅ JSON 解析错误 → 友好错误提示
• ✅ 参数验证 → Pydantic 模型验证

常见时间范围

常见字段

• @timestamp - 日志时间戳
• log - 日志内容
• kubernetes.container_name - Kubernetes 容器名（服务名）
• kubernetes.namespace_name - Kubernetes 命名空间
• kubernetes.pod_name - Pod 名称
• stream - 输出流（stdout/stderr）

故障排查

问题: "Kibana 凭证未配置"

解决: 在 MCP 配置文件（如 mcp_config.json）的 env 部分设置 KIBANA_USERNAME 和 KIBANA_PASSWORD。

问题: "登录失败: 401"

解决: 检查环境变量中的用户名和密码是否正确

问题: "请求超时"

解决:

1. 检查网络连接
2. 减小查询范围（size 参数）
3. 缩短时间范围

问题: 查询返回空结果

解决:

1. 检查索引模式是否正确
2. 扩大时间范围
3. 检查查询语法

开发

运行测试

python server.py

调试模式

设置环境变量启用详细日志:

export DEBUG=1
python server.py

许可证

MIT License