MCP Tools for Elasticsearch — search queries, cluster health, and index management

Elasticsearch REST API architecture for MCP tool authors

Elasticsearch's REST API runs on port 9200 by default. Each node in the cluster exposes the full API — you can send requests to any node and it routes internally. Elastic Cloud and Elastic Cloud on Kubernetes (ECK) expose the same API behind a load balancer with TLS and basic auth or API key authentication. The Authorization: ApiKey <base64(id:api_key)> header or Authorization: Basic <base64(user:pass)> header work for both self-hosted and cloud.

MCP tools should authenticate using an API key with the minimum required cluster and index privileges. A read-only search tool needs only the read index privilege on the target indices. A monitoring tool additionally needs the monitor cluster privilege. Never use the elastic superuser in an MCP tool — key rotation or compromise would have unlimited blast radius.

async function createElasticsearchClient(config) {
  const headers = {
    'Content-Type': 'application/json',
  };

  if (config.apiKey) {
    // Preferred: Elasticsearch API key (id:api_key, base64-encoded)
    const encoded = Buffer.from(`${config.apiKeyId}:${config.apiKey}`).toString('base64');
    headers['Authorization'] = `ApiKey ${encoded}`;
  } else if (config.username && config.password) {
    // Basic auth (self-hosted, dev clusters)
    const encoded = Buffer.from(`${config.username}:${config.password}`).toString('base64');
    headers['Authorization'] = `Basic ${encoded}`;
  }

  async function request(method, path, body) {
    const url = `${config.baseUrl}${path}`;
    const response = await fetch(url, {
      method,
      headers,
      body: body ? JSON.stringify(body) : undefined,
    });

    if (!response.ok) {
      const error = await response.json().catch(() => ({ error: { reason: response.statusText } }));
      throw new Error(
        `Elasticsearch ${method} ${path} → ${response.status}: ${error.error?.reason ?? response.statusText}`
      );
    }

    return response.json();
  }

  return { request };
}

// Usage
const es = await createElasticsearchClient({
  baseUrl: 'https://my-cluster.es.io:9243',
  apiKeyId: 'VuaCfGcBCdbkQm-e5aOx',
  apiKey: 'ui2lp2axTNmsyakw9tvNnw',
});

Cluster health and shard assignment

The GET /_cluster/health endpoint returns the cluster's overall status as green, yellow, or red. Green means all primary and replica shards are assigned. Yellow means all primary shards are assigned but one or more replicas are unassigned — the cluster can serve all reads and writes, but a node loss would result in data being unavailable until replicas are recovered. Red means at least one primary shard is unassigned — documents on that shard are inaccessible for both reads and writes.

A common source of confusion: cluster health is an aggregate across all indices. A single red index drives the whole cluster to red even if all other indices are fully healthy. When debugging red status, always check GET /_cluster/health?level=indices to isolate which index is red, then GET /_cluster/allocation/explain to understand why a shard is unassigned. Common causes: disk watermark exceeded (Elasticsearch stops allocating shards when disk usage passes 85%), node count below the replica minimum, or corrupt shard data requiring a force-allocate.

async function checkClusterHealth(es) {
  const health = await es.request('GET', '/_cluster/health?level=indices&timeout=5s');

  const indexSummary = Object.entries(health.indices ?? {}).map(([name, idx]) => ({
    name,
    status: idx.status,         // 'green' | 'yellow' | 'red'
    activePrimaryShards: idx.active_primary_shards,
    activeShards: idx.active_shards,
    relocatingShards: idx.relocating_shards,
    initializingShards: idx.initializing_shards,
    unassignedShards: idx.unassigned_shards,
  }));

  const redIndices = indexSummary.filter((i) => i.status === 'red');
  const yellowIndices = indexSummary.filter((i) => i.status === 'yellow');

  return {
    clusterName: health.cluster_name,
    status: health.status,
    timedOut: health.timed_out,
    numberOfNodes: health.number_of_nodes,
    numberOfDataNodes: health.number_of_data_nodes,
    activeShards: health.active_shards,
    unassignedShards: health.unassigned_shards,
    activeShardsPercent: health.active_shards_percent_as_number,
    // Actionable summaries
    redIndices,
    yellowIndices,
    healthy: health.status === 'green',
    degraded: health.status === 'yellow',
    critical: health.status === 'red',
  };
}

// When red: explain why shards are unassigned
async function explainUnassignedShards(es) {
  const explanation = await es.request('GET', '/_cluster/allocation/explain');
  return {
    index: explanation.index,
    shard: explanation.shard,
    primary: explanation.primary,
    currentState: explanation.current_state,
    unassignedInfo: explanation.unassigned_info,
    // canAllocate: 'yes' | 'no' | 'throttled' | 'awaiting_info'
    canAllocate: explanation.can_allocate,
    allocateExplanation: explanation.allocate_explanation,
    nodeDecisions: (explanation.node_allocation_decisions ?? []).map((n) => ({
      nodeName: n.node_name,
      decisionType: n.deciders?.[0]?.decision,
      reason: n.deciders?.[0]?.explanation,
    })),
  };
}

Search with Query DSL: bool, must, filter, and should

The Elasticsearch Query DSL structures search requests as a JSON tree. The most important concept for MCP tool authors is the distinction between query context and filter context. In query context, the clause contributes to the relevance _score — Elasticsearch computes how well the document matches. In filter context, the clause is a binary yes/no decision with no score contribution, and results are cached by the field cache for subsequent requests with the same filter.

The bool query combines clauses: must (query context, required, affects score), filter (filter context, required, no score), should (query context, optional — in a bool without must, at least one should must match), must_not (filter context, documents matching this are excluded). For MCP tools that filter by date range, status, or category and then rank by relevance text, always put date/status filters in the filter array and text matching in must with a match or multi_match clause.

async function searchDocuments(es, params) {
  const {
    index,
    text,            // free text to search
    fields = ['title', 'content', 'tags'],
    statusFilter,    // exact value filter (no scoring)
    dateFrom,        // ISO date string
    dateTo,
    size = 20,
    searchAfter,     // for pagination (see next section)
  } = params;

  const mustClauses = [];
  const filterClauses = [];

  if (text) {
    mustClauses.push({
      multi_match: {
        query: text,
        fields,
        type: 'best_fields',
        fuzziness: 'AUTO',
      },
    });
  }

  if (statusFilter) {
    // Exact match — goes in filter context (cached, no score impact)
    filterClauses.push({ term: { status: statusFilter } });
  }

  if (dateFrom || dateTo) {
    filterClauses.push({
      range: {
        created_at: {
          ...(dateFrom && { gte: dateFrom }),
          ...(dateTo && { lte: dateTo }),
          format: 'strict_date_optional_time',
        },
      },
    });
  }

  const query =
    mustClauses.length === 0 && filterClauses.length === 0
      ? { match_all: {} }
      : {
          bool: {
            ...(mustClauses.length > 0 && { must: mustClauses }),
            ...(filterClauses.length > 0 && { filter: filterClauses }),
          },
        };

  const body = {
    query,
    size,
    sort: [{ _score: 'desc' }, { created_at: 'desc' }, { _id: 'asc' }],
    ...(searchAfter && { search_after: searchAfter }),
    _source: ['title', 'status', 'created_at', 'author'],
  };

  const result = await es.request('POST', `/${index}/_search`, body);

  return {
    total: result.hits.total.value,
    totalRelation: result.hits.total.relation, // 'eq' or 'gte' (when > 10,000)
    hits: result.hits.hits.map((h) => ({
      id: h._id,
      score: h._score,
      source: h._source,
      sort: h.sort, // use as next search_after value
    })),
  };
}

Pagination with search_after and point-in-time

Elasticsearch limits from + size pagination to 10,000 total hits by default (the index.max_result_window setting). Requests beyond this throw a Result window is too large error. The scroll API was the historical workaround but is deprecated for deep pagination in Elasticsearch 7.10+ — it holds a consistent view of the index but consumes significant heap memory on the coordinating node and does not support sorting by relevance.

The correct modern pattern is search_after with a Point in Time (PIT). A PIT freezes a lightweight logical snapshot of the index state at the moment it is opened, allowing you to paginate through a consistent result set even as the underlying index receives new writes. PITs are cheap to open (they reference existing Lucene segment readers) but must be explicitly deleted when done — they hold open file descriptors. PITs expire after the configured keep-alive, so refresh the keep-alive on each paginated request by including the PIT id in the search request body.

async function paginatedSearch(es, index, query, pageSize = 100) {
  // 1. Open a Point in Time — holds a consistent snapshot of the index
  const pitResponse = await es.request('POST', `/${index}/_pit?keep_alive=5m`);
  const pitId = pitResponse.id;

  const allResults = [];
  let searchAfter = undefined;

  try {
    while (true) {
      const body = {
        query,
        size: pageSize,
        sort: [{ created_at: 'desc' }, { _id: 'asc' }],
        pit: {
          id: pitId,
          keep_alive: '5m', // Refreshes the PIT keep-alive on each request
        },
        ...(searchAfter && { search_after: searchAfter }),
        // Do not specify index in the URL when using PIT — it's encoded in the PIT
        track_total_hits: false, // Faster when you don't need total count
      };

      // Note: when using PIT, send to /_search not /{index}/_search
      const result = await es.request('POST', '/_search', body);
      const hits = result.hits.hits;

      if (hits.length === 0) break;

      allResults.push(...hits.map((h) => ({ id: h._id, source: h._source })));
      searchAfter = hits[hits.length - 1].sort;

      if (hits.length < pageSize) break; // Last page
    }
  } finally {
    // Always delete the PIT — leaking PITs exhausts file descriptors
    await es.request('DELETE', '/_pit', { id: pitId }).catch(() => {});
  }

  return allResults;
}

// For a single next-page request (stateless pagination, no PIT):
async function getNextPage(es, index, query, sortValues, pageSize = 20) {
  const body = {
    query,
    size: pageSize,
    sort: [{ created_at: 'desc' }, { _id: 'asc' }],
    search_after: sortValues, // Pass the `sort` array from the last hit of the previous page
  };
  const result = await es.request('POST', `/${index}/_search`, body);
  return {
    hits: result.hits.hits,
    nextSortValues: result.hits.hits.at(-1)?.sort ?? null,
  };
}

Aggregations: counts, date histograms, and terms

Aggregations compute summaries over a result set — counts, sums, averages, top values, date histograms. They run alongside the search query and share the same filter context. For MCP tools that power dashboards or analytics, aggregations are more efficient than fetching and processing individual documents: a date_histogram over 10 million documents returns a bucketed summary in milliseconds compared to paginating through all results.

Aggregations have a size: 0 optimization: setting size to zero in the search body skips returning individual hits and only computes aggregations. Use this when you need summaries, not documents. The terms aggregation has a default size: 10 that limits results to the top 10 buckets — always set an explicit size if you need more unique values, and check the sum_other_doc_count field to know how many documents fell outside the returned buckets.

async function getIndexAggregations(es, index, dateField, statusField, fromDate, toDate) {
  const body = {
    size: 0, // No individual hits — only aggregations
    query: {
      bool: {
        filter: [
          { range: { [dateField]: { gte: fromDate, lte: toDate } } },
        ],
      },
    },
    aggs: {
      // Count documents by day
      events_over_time: {
        date_histogram: {
          field: dateField,
          calendar_interval: 'day',
          format: 'yyyy-MM-dd',
          min_doc_count: 0, // Include empty buckets
          extended_bounds: { min: fromDate, max: toDate },
        },
      },
      // Count documents by status value (top 20 values)
      by_status: {
        terms: {
          field: `${statusField}.keyword`, // .keyword suffix for exact-match aggregations
          size: 20,
          order: { _count: 'desc' },
          missing: '__missing__', // Count docs where statusField is absent
        },
      },
      // Percentile response times (for latency monitoring)
      latency_percentiles: {
        percentiles: {
          field: 'response_time_ms',
          percents: [50, 75, 90, 95, 99],
        },
      },
    },
  };

  const result = await es.request('POST', `/${index}/_search`, body);
  const aggs = result.aggregations;

  return {
    totalInRange: result.hits.total.value,
    dailyBuckets: aggs.events_over_time.buckets.map((b) => ({
      date: b.key_as_string,
      count: b.doc_count,
    })),
    statusCounts: aggs.by_status.buckets.map((b) => ({
      status: b.key,
      count: b.doc_count,
    })),
    otherStatusCount: aggs.by_status.sum_other_doc_count,
    latencyPercentiles: aggs.latency_percentiles.values,
  };
}

Index health and lifecycle management

Beyond cluster health, MCP tools managing Elasticsearch infrastructure often need to inspect index-level metadata: document counts, store sizes, mapping details, and Index Lifecycle Management (ILM) state. The GET /_cat/indices and GET /{index}/_stats endpoints provide this data. For clusters using ILM (time-series data like logs, metrics, traces), the GET /{index}/_ilm/explain endpoint shows which ILM phase an index is in and why it may be stuck.

A common ILM pitfall: an index stuck in the delete phase because ILM is trying to delete an index that is the write index for a data stream. ILM cannot delete a data stream's write index — you must roll over the data stream first. The step.name and step_info.reason fields in the ILM explain response identify this condition.

async function getIndexDetails(es, indexPattern) {
  // Cat API for a quick summary — use v for column headers, bytes for human sizes
  const catResult = await es.request(
    'GET',
    `/_cat/indices/${indexPattern}?v&h=index,health,status,docs.count,store.size,pri,rep&s=index&format=json`
  );

  // Detailed stats for a specific index
  async function getIndexStats(index) {
    const stats = await es.request('GET', `/${index}/_stats/docs,store`);
    const primaries = stats._all.primaries;
    const total = stats._all.total;
    return {
      docsCount: primaries.docs.count,
      docsDeleted: primaries.docs.deleted,
      primaryStoreBytes: primaries.store.size_in_bytes,
      totalStoreBytes: total.store.size_in_bytes, // Primary + replica
    };
  }

  // ILM status for an index
  async function getILMStatus(index) {
    try {
      const ilm = await es.request('GET', `/${index}/_ilm/explain`);
      const indexInfo = ilm.indices[index];
      if (!indexInfo) return { managed: false };

      return {
        managed: indexInfo.managed,
        policy: indexInfo.policy,
        phase: indexInfo.phase,       // 'hot' | 'warm' | 'cold' | 'delete'
        action: indexInfo.action,     // e.g. 'rollover', 'shrink', 'delete'
        step: indexInfo.step,         // current step within the action
        stepTime: indexInfo.step_time_millis,
        // If stuck: phase_execution.phase_definition shows what triggered
        stuck: indexInfo.failed_step !== undefined,
        failedStep: indexInfo.failed_step,
        stepInfo: indexInfo.step_info, // Error details if stuck
      };
    } catch {
      return { managed: false };
    }
  }

  return {
    indices: catResult,
    getIndexStats,
    getILMStatus,
  };
}

AliveMCP integration for Elasticsearch health monitoring

Elasticsearch cluster health is a multi-layer condition: the cluster can be green while individual indices are yellow, while ILM is stuck on some indices, and while the JVM heap is above 75% (which triggers excessive GC and eventual circuit-breaker errors). A complete Elasticsearch health probe covers all three layers independently.

Register a custom HTTP health endpoint with AliveMCP that checks cluster status, JVM heap, and disk watermarks. The Elasticsearch GET /_nodes/stats/jvm,fs endpoint returns per-node JVM heap usage and filesystem statistics. Alert on heap above 75% (long GC pauses begin) and disk above 85% (Elasticsearch's default low watermark where shard allocation stops). Set probe interval to 60 seconds — Elasticsearch cluster stats update every 30 seconds by default, so sub-30s polling returns stale data.

async function elasticsearchHealthProbe(es) {
  const [clusterHealth, nodeStats] = await Promise.all([
    es.request('GET', '/_cluster/health'),
    es.request('GET', '/_nodes/stats/jvm,fs'),
  ]);

  const nodeDetails = Object.values(nodeStats.nodes).map((node) => {
    const heapUsedPercent = node.jvm.mem.heap_used_percent;
    const diskTotalBytes = node.fs.total.total_in_bytes;
    const diskAvailBytes = node.fs.total.available_in_bytes;
    const diskUsedPercent = Math.round(
      ((diskTotalBytes - diskAvailBytes) / diskTotalBytes) * 100
    );

    return {
      name: node.name,
      roles: node.roles,
      heapUsedPercent,
      diskUsedPercent,
      jvmWarning: heapUsedPercent > 75,
      diskLowWatermark: diskUsedPercent > 85,
      diskHighWatermark: diskUsedPercent > 90,
      diskFloodStage: diskUsedPercent > 95,
    };
  });

  const worstHeap = Math.max(...nodeDetails.map((n) => n.heapUsedPercent));
  const worstDisk = Math.max(...nodeDetails.map((n) => n.diskUsedPercent));

  return {
    clusterStatus: clusterHealth.status,
    numberOfNodes: clusterHealth.number_of_nodes,
    unassignedShards: clusterHealth.unassigned_shards,
    // true = cluster can serve reads and writes for all primary shards
    primaryShardsHealthy: clusterHealth.status !== 'red',
    jvmWarning: worstHeap > 75,
    diskLowWatermark: worstDisk > 85,
    nodeDetails,
    // Composite health: all three must be clear for green
    healthy: clusterHealth.status === 'green' && worstHeap <= 75 && worstDisk <= 85,
  };
}

Register this endpoint with AliveMCP using a 60-second probe interval and a 10-second timeout (node stats collection can be slow on large clusters). Configure a separate AliveMCP check for each critical index using the index health endpoint — a single green cluster-level check will not alert you when a specific index turns red.

Related guides

• MCP tools for Grafana Loki — LogQL queries, log push, and label streaming
• MCP tools for Prometheus Alertmanager — alerts, silences, and inhibition rules
• Distributed tracing for MCP servers — trace context propagation and sampling
• MCP server observability — metrics, logs, and traces for production MCP tools
• MCP server health check patterns — composite endpoints and failure classification
• Structured logging for MCP servers — JSON fields, correlation IDs, and log levels