MCP Tools API Reference

Kafka Log Analyzer MCP Tools 完整 API 文档

概述

Kafka Log Analyzer 通过 MCP (Model Context Protocol) 暴露以下工具，可从 Claude Code 或任何 MCP 客户端调用。

---

Tools

analyze_log

解析 Kafka 日志，提取事件并检测异常。

输入 Schema

{
  source: 'paste' | 'file' | 'exporter' | 'loki',  // 数据源类型
  content?: string,                  // 日志内容（source=paste 时必填）
  path?: string,                     // 文件路径（source=file 时必填）
  cluster?: string,                  // 集群名称（exporter/loki 源时使用）
  query?: string,                    // LogQL 查询表达式（loki 源时使用）
  limit?: number,                    // 最大日志行数（loki 源时默认1000）
  focus?: FocusArea[],               // 关注点过滤
  timeline?: TimelineWindow,         // 时间线窗口
  priority?: Priority[],             // 按优先级过滤异常（P0-P3）
  report?: 'markdown' | 'json' | 'slack' | 'folded-markdown'  // 输出格式（默认 json）
}

过滤语义

• focus — 组件过滤（producer/consumer/broker）在 Python 层执行；语义过滤（lag 过滤 consumer_lag 事件，error 过滤 ERROR 级别事件及 P0/P1 异常）在 TS 层执行。两者可组合使用。
• priority — 仅过滤异常（事件本身没有优先级）。在 severity 映射之后应用。
• report — 当 json（默认）时，返回结构化的 AnalyzeLogOutput；当 markdown/slack 时，通过 formatReport() 返回格式化文本。

输出

{
  events: Event[],                   // 提取的事件列表
  anomalies: Anomaly[],              // 检测到的异常
  summary: {                         // 分析摘要
    total: number,
    byPriority: {
      P0: number,
      P1: number,
      P2: number,
      P3: number
    },
    byComponent: {
      producer: number,
      consumer: number,
      broker: number
    }
  },
  timeline?: TimelineBucket[]        // 时间线分布（当指定 timeline 时）
}

事件类型定义

interface Event {
  timestamp: string;                 // 事件时间戳
  level: 'INFO' | 'WARN' | 'ERROR' | 'FATAL';
  component: 'producer' | 'consumer' | 'broker';
  type: EventType;                   // 事件类型
  message: string;                   // 原始消息
  priority: 'P0' | 'P1' | 'P2' | 'P3';
  metadata?: Record<string, unknown>;
}

type EventType =
  | 'send_success'
  | 'send_failure'
  | 'consumer_lag'
  | 'rebalance'
  | 'commit_failure'
  | 'buffer_exhausted'
  | 'leader_change'
  | 'offset_out_of_range'
  | 'serialization_error'
  | 'network_error'
  | 'auth_error';

异常类型定义

interface Anomaly {
  type: AnomalyType;                 // 异常类型
  severity: 'P0' | 'P1' | 'P2' | 'P3';
  component: 'producer' | 'consumer' | 'broker';
  description: string;               // 异常描述
  recommendation: string;            // 修复建议
  affectedEvents: number;            // 影响的事件数
  metadata?: Record<string, unknown>;
}

type AnomalyType =
  | 'error_rate_spike'               // 错误率突增
  | 'rebalance_storm'                // Rebalance 风暴
  | 'lag_spike'                      // 消费积压突增
  | 'leader_instability'             // Leader 频繁切换
  | 'replica_lag'                    // 副本同步延迟
  | 'serialization_issue'            // 序列化问题
  | 'network_problem';               // 网络异常

调用示例

粘贴日志分析：

{
  "source": "paste",
  "content": "[2026-01-15 10:00:01] ERROR [producer] Failed to send record to topic orders\n[2026-01-15 10:00:02] WARN  [consumer] lag exceeded threshold (5000 messages)",
  "focus": ["producer", "error"],
  "timeline": "1h"
}

文件日志分析：

{
  "source": "file",
  "path": "/var/log/kafka/server.log",
  "focus": ["consumer", "lag"],
  "timeline": "15m"
}

---

get_lag

从 Kafka Exporter / Prometheus 获取 Consumer Lag 指标。

输入 Schema

{
  cluster?: string,                  // 集群名称（可选，默认所有集群）
  consumer_group?: string,           // 消费组名称（可选）
  topic?: string                     // Topic 名称（可选）
}

输出

{
  lags: LagEntry[],                  // Lag 数据列表
  timestamp: string                  // 查询时间戳
}

interface LagEntry {
  cluster: string;                   // 集群名称
  group: string;                     // 消费组
  topic: string;                     // Topic
  partition: number;                 // 分区号
  currentOffset: number;             // 当前偏移量
  endOffset: number;                 // 末端偏移量
  lag: number;                       // 积压数量
  timestamp: string;                 // 数据时间戳
}

调用示例

查询所有消费组 Lag：

{}

查询指定集群和消费组：

{
  "cluster": "production",
  "consumer_group": "order-processor"
}

查询指定 Topic：

{
  "topic": "orders"
}

---

start_hooks

启动 Hook HTTP 服务器，接收 Grafana/PagerDuty 告警 Webhook。

输入 Schema

{
  port?: number  // 监听端口（默认：GRAFANA_WEBHOOK_PORT 或 3100）
}

输出

{
  status: 'started',
  port: number,
  endpoints: {
    grafana: '/hooks/grafana',
    pagerduty: '/hooks/pagerduty',
    health: '/hooks/health'
  }
}

调用示例

{
  "tool": "start_hooks",
  "input": {
    "port": 3100
  }
}

---

stop_hooks

停止 Hook HTTP 服务器。

输入 Schema

{}  // 无参数

输出

{
  status: 'stopped'
}

---

list_hooks

查询 Hook 服务器状态和去重统计。

输入 Schema

{}  // 无参数

输出

{
  running: boolean,
  port?: number,
  dedupStats: {
    activeEntries: number,
    totalMerged: number,
    totalExpired: number
  }
}

---

query_history (v0.5.0+)

查询历史分析记录，支持多条件过滤。

输入 Schema

{
  from?: string,                       // 起始时间 ISO 8601（默认：7 天前）
  to?: string,                         // 结束时间 ISO 8601（默认：当前）
  source?: 'paste' | 'file' | 'exporter' | 'loki',  // 按数据源过滤
  cluster?: string,                    // 按集群过滤
  limit?: number                       // 结果数量（默认 50，最大 200）
}

输出

AnalysisRecord[]  // 历史分析记录列表

调用示例

查询最近 24 小时的分析记录：

{
  "from": "2026-07-04T00:00:00Z",
  "to": "2026-07-05T00:00:00Z"
}

---

compare_trend (v0.5.0+)

比较两个时间范围内的指标变化趋势。

输入 Schema

{
  metric: 'error_rate' | 'lag' | 'anomaly_count' | 'event_count',
  current_range: '1h' | '6h' | '1d' | '7d',
  compare_range?: 'previous' | 'last_week' | 'last_month',  // 默认：previous
  cluster?: string
}

输出

interface TrendComparison {
  metric: string;
  current: { range: string; avg: number; max: number };
  compare: { range: string; avg: number; max: number };
  change: { avgMultiplier: number; maxMultiplier: number; trend: 'up' | 'down' | 'stable' };
  summary: string;
}

调用示例

{
  "metric": "error_rate",
  "current_range": "1h",
  "compare_range": "previous"
}

---

set_baseline (v0.5.0+)

手动设置指标基线，覆盖自动基线。

输入 Schema

{
  metric_key: string,                  // 例如 "lag:cluster:group:topic"
  value: number,                       // 基线值
  metric_type: 'lag' | 'error_rate' | 'anomaly_count'
}

调用示例

{
  "metric_key": "lag:prod:order-processor:orders",
  "value": 5000,
  "metric_type": "lag"
}

---

list_baselines (v0.5.0+)

列出所有基线，可选按指标类型过滤。

输入 Schema

{
  metric_type?: 'lag' | 'error_rate' | 'anomaly_count'
}

调用示例

列出所有 Lag 基线：

{
  "metric_type": "lag"
}

---

cleanup_storage (v0.5.0+)

清理过期的持久化数据。

输入 Schema

{
  retention_days?: number,             // 保留天数（默认 30）
  dry_run?: boolean                    // 仅预览，不删除
}

调用示例

预览清理结果：

{
  "retention_days": 30,
  "dry_run": true
}

---

diagnose (v1.1.0+)

使用诊断模板排查常见 Kafka 问题。

输入 Schema

{
  template_id: 'lag-diagnosis' | 'rebalance-storm' | 'producer-errors' | 'broker-health' | 'full-audit',
  source: 'paste' | 'file' | 'exporter' | 'loki',
  content?: string,                  // 日志内容（source=paste 时使用）
  path?: string,                     // 文件路径（source=file 时使用）
  cluster?: string,                  // 集群名称（source=exporter/loki 时使用）
  report?: 'json' | 'markdown' | 'folded-markdown'  // 输出格式（默认 json）
}

输出

返回应用了模板焦点和优先级设置的分析结果。模板的 focus 和 priority 默认值与提供的 source 参数合并（CLI 标志优先）。

调用示例

{
  "tool": "diagnose",
  "input": {
    "template_id": "lag-diagnosis",
    "source": "file",
    "path": "/var/log/kafka/server.log",
    "report": "folded-markdown"
  }
}

---

list_templates (v1.1.0+)

列出可用诊断模板及推荐评分。

输入 Schema

{
  cluster?: string  // 集群名称，用于个性化推荐（可选）
}

输出

[
  {
    id: string,
    name: string,
    description: string,
    focus_hints: string[]
  }
]

模板按指定集群的历史异常频率排序。无集群参数或无历史记录时，返回默认排序的全部模板。

调用示例

{
  "tool": "list_templates",
  "input": {
    "cluster": "production"
  }
}

---

timeline

按时间窗口统计事件分布。

输入 Schema

{
  events: Event[],                   // 事件列表
  window: TimelineWindow             // 时间窗口
}

type TimelineWindow = '1m' | '5m' | '15m' | '1h' | '6h' | '1d';

输出

{
  buckets: TimelineBucket[],         // 时间桶列表
  window: TimelineWindow,            // 使用的窗口大小
  totalBuckets: number               // 总桶数
}

interface TimelineBucket {
  start: string;                     // 桶起始时间
  end: string;                       // 桶结束时间
  count: number;                     // 事件数
  byLevel: {                         // 按级别统计
    INFO: number;
    WARN: number;
    ERROR: number;
    FATAL: number;
  },
  byComponent: {                     // 按组件统计
    producer: number;
    consumer: number;
    broker: number;
  }
}

调用示例

{
  "events": [
    { "timestamp": "2026-01-15 10:00:01", "level": "ERROR", "component": "producer", "type": "send_failure", "message": "Failed to send", "priority": "P1" }
  ],
  "window": "5m"
}

---

Resources

kafka://metrics/{cluster}

Consumer Lag 实时指标流。

访问方式

// 通过 MCP Resource 订阅
const resource = await client.readResource({
  uri: 'kafka://metrics/production/lag'
});

数据格式

{
  cluster: string;
  groups: {
    name: string;
    totalLag: number;
    partitions: {
      topic: string;
      partition: number;
      lag: number;
    }[];
  }[];
  timestamp: string;
}

---

kafka://history/{cluster}

历史分析记录。

Phase 4 功能

访问方式

const resource = await client.readResource({
  uri: 'kafka://history/production'
});

---

错误处理

错误响应格式

{
  error: {
    code: string;                    // 错误码
    message: string;                 // 错误信息
    details?: Record<string, unknown>;
  }
}

常见错误码

错误码	HTTP 等价	说明
INVALID_INPUT	400	输入参数验证失败 → McpError(InvalidParams)
FILE_NOT_FOUND	404	指定文件不存在 → McpError(InvalidParams)
PARSE_ERROR	422	日志解析失败 → McpError(InternalError)
PROMETHEUS_UNAVAILABLE	503	Prometheus 连接失败
LOKI_UNAVAILABLE	503	Loki 连接失败
INTERNAL_ERROR	500	内部错误 → McpError(InternalError)

错误处理示例

try {
  const result = await analyzeLog({ source: 'file', path: '/nonexistent.log' });
} catch (error) {
  if (error.code === 'FILE_NOT_FOUND') {
    console.error('File not found:', error.message);
  }
}

---

优先级规则

P0 (Critical)

集群宕机、数据丢失风险、完全不可用。

P1 (High)

Consumer Lag > 10K、频繁 Rebalance、持续错误率 > 5%。

P2 (Medium)

Leader 切换、瞬时错误、配置告警。

P3 (Low)

警告信息、通知事件、统计信息。

---

相关文档：

• 架构概览
• 配置指南
• 命令参考