MCP Tools API Reference

Complete API documentation for Kafka Log Analyzer MCP Tools

Overview

Kafka Log Analyzer exposes the following tools via MCP (Model Context Protocol), callable from Claude Code or any MCP client.

---

Tools

analyze_log

Parse Kafka logs, extract events, and detect anomalies.

Input Schema

{
  source: 'paste' | 'file' | 'exporter' | 'loki',  // Data source type
  content?: string,                  // Log content (required when source=paste)
  path?: string,                     // File path (required when source=file)
  cluster?: string,                  // Cluster name (for exporter/loki sources)
  query?: string,                    // LogQL query (for loki source)
  limit?: number,                    // Max log lines for loki (default: 1000)
  focus?: FocusArea[],               // Focus area filter
  timeline?: TimelineWindow,         // Timeline window
  priority?: Priority[],             // Filter anomalies by priority (P0-P3)
  report?: 'markdown' | 'json' | 'slack' | 'folded-markdown'  // Output format (default: json)
}

type FocusArea = 'producer' | 'consumer' | 'broker' | 'lag' | 'error';
type Priority = 'P0' | 'P1' | 'P2' | 'P3';
type TimelineWindow = '1m' | '5m' | '15m' | '1h' | '6h' | '1d';

Filter Semantics

• focus — Component filter (producer/consumer/broker) is applied in Python; semantic filter (lag filters consumer_lag events, error filters ERROR-level events and P0/P1 anomalies) is applied in TS. Both can be combined.
• priority — Only filters anomalies (events have no priority). Applied after severity mapping.
• report — When json (default), returns structured AnalyzeLogOutput. When markdown/slack, returns formatted text via formatReport().

Output

{
  events: Event[],                   // Extracted events list
  anomalies: Anomaly[],              // Detected anomalies
  summary: {                         // Analysis summary
    total: number,
    byPriority: {
      P0: number,
      P1: number,
      P2: number,
      P3: number
    },
    byComponent: {
      producer: number,
      consumer: number,
      broker: number
    }
  },
  timeline?: TimelineBucket[]        // Timeline distribution (when timeline specified)
}

Event Type Definition

interface Event {
  timestamp: string;                 // Event timestamp
  level: 'INFO' | 'WARN' | 'ERROR' | 'FATAL';
  component: 'producer' | 'consumer' | 'broker';
  type: EventType;                   // Event type
  message: string;                   // Original message
  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';

Anomaly Type Definition

interface Anomaly {
  type: AnomalyType;                 // Anomaly type
  severity: 'P0' | 'P1' | 'P2' | 'P3';
  component: 'producer' | 'consumer' | 'broker';
  description: string;               // Anomaly description
  recommendation: string;            // Fix recommendation
  affectedEvents: number;            // Affected event count
  metadata?: Record<string, unknown>;
}

type AnomalyType =
  | 'error_rate_spike'               // Error rate spike
  | 'rebalance_storm'                // Rebalance storm
  | 'lag_spike'                      // Consumer lag spike
  | 'leader_instability'             // Leader frequent change
  | 'replica_lag'                    // Replica sync lag
  | 'serialization_issue'            // Serialization issue
  | 'network_problem';               // Network anomaly

Call Examples

Paste Log Analysis:

{
  "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"
}

File Log Analysis:

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

---

get_lag

Get Consumer Lag metrics from Kafka Exporter / Prometheus.

Input Schema

{
  cluster?: string,                  // Cluster name (optional, default all clusters)
  consumer_group?: string,           // Consumer group name (optional)
  topic?: string                     // Topic name (optional)
}

Output

{
  lags: LagEntry[],                  // Lag data list
  timestamp: string,                 // Query timestamp
  warning?: string                   // Warning when source is unavailable (degraded mode)
}

interface LagEntry {
  cluster: string;                   // Cluster name
  group: string;                     // Consumer group
  topic: string;                     // Topic
  partition: number;                 // Partition number
  currentOffset: number;             // Current offset
  endOffset: number;                 // End offset
  lag: number;                       // Lag count
  timestamp: string;                 // Data timestamp
}

Call Examples

Query All Consumer Groups Lag:

{}

Query Specific Cluster and Consumer Group:

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

Query Specific Topic:

{
  "topic": "orders"
}

---

start_hooks

Start the hook HTTP server for receiving Grafana/PagerDuty alert webhooks.

Input Schema

{
  port?: number  // Listen port (default: GRAFANA_WEBHOOK_PORT or 3100)
}

Output

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

Call Example

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

---

stop_hooks

Stop the hook HTTP server.

Input Schema

{}  // No parameters

Output

{
  status: 'stopped'
}

---

list_hooks

List hook server status and dedup statistics.

Input Schema

{}  // No parameters

Output

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

---

query_history (v0.5.0+)

Query historical analysis records with filters.

Input Schema

{
  from?: string,                       // Start time ISO 8601 (default: 7 days ago)
  to?: string,                         // End time ISO 8601 (default: now)
  source?: 'paste' | 'file' | 'exporter' | 'loki',  // Filter by source
  cluster?: string,                    // Filter by cluster
  limit?: number                       // Result count (default: 50, max: 200)
}

Output

AnalysisRecord[]  // List of historical analysis records

---

compare_trend (v0.5.0+)

Compare a metric across two time ranges.

Input Schema

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

Output

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;
}

---

set_baseline (v0.5.0+)

Manually set a metric baseline, overriding auto baselines.

Input Schema

{
  metric_key: string,                  // e.g. "lag:cluster:group:topic"
  value: number,                       // Baseline value
  metric_type: 'lag' | 'error_rate' | 'anomaly_count'
}

---

list_baselines (v0.5.0+)

List all baselines, optionally filtered by metric type.

Input Schema

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

---

cleanup_storage (v0.5.0+)

Clean up expired persisted data.

Input Schema

{
  retention_days?: number,             // Retention in days (default: 30)
  dry_run?: boolean                    // Preview only, no deletion
}

---

diagnose (v1.1.0+)

Run a diagnostic template for common Kafka issues.

Input Schema

{
  template_id: 'lag-diagnosis' | 'rebalance-storm' | 'producer-errors' | 'broker-health' | 'full-audit',
  source: 'paste' | 'file' | 'exporter' | 'loki',
  content?: string,                  // Log content (when source=paste)
  path?: string,                     // File path (when source=file)
  cluster?: string,                  // Cluster name (when source=exporter/loki)
  report?: 'json' | 'markdown' | 'folded-markdown'  // Output format (default: json)
}

Output

Returns the analysis result with the template's focus and priority settings pre-applied. The template's focus and priority defaults merge with the provided source parameters (CLI flags take precedence).

Call Example

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

---

list_templates (v1.1.0+)

List available diagnostic templates with recommendation scores.

Input Schema

{
  cluster?: string  // Cluster name for personalized recommendations (optional)
}

Output

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

Templates are ordered by relevance based on historical anomaly patterns for the given cluster. If no cluster is specified or no history exists, returns all templates in default order.

Call Example

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

---

timeline

Statistical event distribution by time window.

Input Schema

{
  events: Event[],                   // Event list
  window: TimelineWindow             // Time window
}

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

Output

{
  buckets: TimelineBucket[],         // Time bucket list
  window: TimelineWindow,            // Window size used
  totalBuckets: number               // Total bucket count
}

interface TimelineBucket {
  start: string;                     // Bucket start time
  end: string;                       // Bucket end time
  count: number;                     // Event count
  byLevel: {                         // By level statistics
    INFO: number;
    WARN: number;
    ERROR: number;
    FATAL: number;
  },
  byComponent: {                     // By component statistics
    producer: number;
    consumer: number;
    broker: number;
  }
}

Call Example

{
  "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 real-time metrics stream.

Access Method

// Subscribe via MCP Resource
const resource = await client.readResource({
  uri: 'kafka://metrics/production/lag'
});

Data Format

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

---

kafka://history/{cluster}

Historical analysis records.

Phase 4 Feature

Access Method

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

---

Error Handling

Error Response Format

{
  error: {
    code: string;                    // Error code
    message: string;                 // Error message
    details?: Record<string, unknown>;
  }
}

Common Error Codes

Error Code	HTTP Equivalent	Description
INVALID_INPUT	400	Input parameter validation failed → McpError(InvalidParams)
FILE_NOT_FOUND	404	Specified file not found → McpError(InvalidParams)
PARSE_ERROR	422	Log parsing failed → McpError(InternalError)
PROMETHEUS_UNAVAILABLE	503	Prometheus connection failed
LOKI_UNAVAILABLE	503	Loki connection failed
DATABASE_ERROR	503	SQLite operation failed
INTERNAL_ERROR	500	Internal error → McpError(InternalError)

Error Handling Example

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);
  }
}

---

Priority Rules

P0 (Critical)

Cluster downtime, data loss risk, completely unavailable.

P1 (High)

Consumer Lag > 10K, frequent Rebalance, sustained error rate > 5%.

P2 (Medium)

Leader switch, transient errors, configuration alerts.

P3 (Low)

Warning messages, notification events, statistical information.

---

Related Documentation:

• Architecture Overview
• Configuration Guide
• Command Reference