Guide · Search Integration

MCP Tools for OpenSearch — bulk indexing, scroll API, AWS SigV4 auth

Three OpenSearch behaviours catch MCP server authors off-guard: the @elastic/elasticsearch v8.x client adds X-Elastic-Client-Meta and Elastic-Api-Version headers that older OpenSearch versions reject with 406 or 400 errors — always use @opensearch-project/opensearch when targeting an OpenSearch cluster, not the Elasticsearch client; the Bulk API always returns HTTP 200 even when individual documents fail to index — you must check response.body.errors === true and iterate response.body.items to find per-document error objects; and AWS OpenSearch Service requires SigV4 request signing — basic-auth credentials used for self-hosted OpenSearch do not work with managed AWS clusters, and the signing logic must be wired through a custom transport or the @opensearch-project/opensearch/aws-connector package.

TL;DR

Use @opensearch-project/opensearch not @elastic/elasticsearch. Check body.errors after every bulk call. Use SigV4 connector for AWS OpenSearch Service. Use scroll API for full result exports; use point-in-time (PIT) for consistent paginated reads. Health probe: GET /_cluster/health.

Client setup — opensearch-js vs elasticsearch client

OpenSearch was forked from Elasticsearch 7.10. The @elastic/elasticsearch v8 client sends headers and uses API patterns that are incompatible with OpenSearch. The @opensearch-project/opensearch client is the correct dependency for any OpenSearch target, including self-hosted OpenSearch and Amazon OpenSearch Service.

// Correct: use the OpenSearch client, not the Elasticsearch client
import { Client } from '@opensearch-project/opensearch';

// Self-hosted OpenSearch with basic auth
const client = new Client({
  node: process.env.OPENSEARCH_URL ?? 'http://localhost:9200',
  auth: {
    username: process.env.OPENSEARCH_USERNAME ?? 'admin',
    password: process.env.OPENSEARCH_PASSWORD!,
  },
  // TLS verification — disable only in local dev
  ssl: {
    rejectUnauthorized: process.env.NODE_ENV === 'production',
  },
  // Retry configuration for transient network errors
  maxRetries: 3,
  requestTimeout: 30_000,  // 30 seconds per request
  sniffOnStart: false,     // Disable node sniffing — incompatible with load balancers
});

// AWS OpenSearch Service — requires SigV4 signing
import { defaultProvider } from '@aws-sdk/credential-provider-node';
import { AwsSigv4Signer } from '@opensearch-project/opensearch/aws';

const awsClient = new Client({
  ...AwsSigv4Signer({
    region: process.env.AWS_REGION ?? 'us-east-1',
    service: 'es',  // 'es' for OpenSearch Service, 'aoss' for Serverless
    getCredentials: () => {
      const credentialsProvider = defaultProvider();
      return credentialsProvider();
    },
  }),
  node: `https://${process.env.OPENSEARCH_DOMAIN_ENDPOINT}`,
  // No auth field needed — SigV4 handles authentication
});

For Amazon OpenSearch Serverless, use service: 'aoss' instead of 'es'. The Serverless API surface is a subset of OpenSearch — it does not support scroll, _cat APIs, or cluster-level admin operations, which affects MCP tools that rely on those endpoints.

Bulk indexing — check per-document errors in the response

OpenSearch's Bulk API processes multiple operations in a single HTTP request. The response always returns HTTP 200 even when individual operations fail (routing failures, version conflicts, mapping errors). You must check the top-level errors field and iterate items to find per-document failures. Logging only the HTTP status code gives a false impression of success.

// MCP tool: bulk index documents with per-document error checking
server.tool('bulk_index_documents', async ({ documents, indexName }) => {
  // Bulk body alternates between action descriptors and document bodies
  const body = documents.flatMap(doc => [
    {
      index: {
        _index: indexName,
        _id: doc.id,                  // Explicit ID — omit for auto-generated
        // version_type: 'external',  // Uncomment for external version control
        // version: doc.version,
      }
    },
    {
      title: doc.title,
      description: doc.description,
      category: doc.category,
      price: doc.price,
      tags: doc.tags,
      indexedAt: new Date().toISOString(),
    },
  ]);

  const response = await client.bulk({ refresh: 'wait_for', body });

  // ALWAYS check the errors field — HTTP 200 does not mean all docs succeeded
  if (response.body.errors) {
    const failed = response.body.items
      .filter((item: any) => item.index?.error)
      .map((item: any) => ({
        id: item.index._id,
        status: item.index.status,
        error: item.index.error,
      }));

    console.warn(`Bulk index: ${failed.length} documents failed out of ${documents.length}`);

    return {
      total: documents.length,
      succeeded: documents.length - failed.length,
      failed: failed.length,
      errors: failed.slice(0, 10), // Return first 10 for diagnosis
    };
  }

  return {
    total: documents.length,
    succeeded: documents.length,
    failed: 0,
  };
});

// MCP tool: bulk upsert using update action with doc_as_upsert
server.tool('bulk_upsert_documents', async ({ documents, indexName }) => {
  const body = documents.flatMap(doc => [
    { update: { _index: indexName, _id: String(doc.id) } },
    {
      doc: { ...doc, updatedAt: new Date().toISOString() },
      doc_as_upsert: true,  // Create if not exists, update if exists
    },
  ]);

  const response = await client.bulk({ refresh: false, body });

  const failed = response.body.errors
    ? response.body.items
        .filter((item: any) => item.update?.error)
        .map((item: any) => ({ id: item.update._id, error: item.update.error }))
    : [];

  return {
    total: documents.length,
    failed: failed.length,
    errors: failed.slice(0, 5),
  };
});

The refresh: 'wait_for' parameter forces OpenSearch to refresh the index shard before returning — indexed documents are immediately searchable after the bulk call. This is slower than refresh: false (async refresh) but avoids stale-read problems in MCP tools that index then immediately search.

Search tool patterns — bool queries and pagination

OpenSearch uses a JSON query DSL. The primary building block is the bool query, which combines must (AND, affects score), filter (AND, no scoring, cached), should (OR with boost), and must_not (NOT) clauses. Always use filter context for exact-match conditions — it is faster than must because OpenSearch caches filter results.

// MCP tool: search with bool query
server.tool('search_documents', async ({
  query,
  category,
  minPrice,
  maxPrice,
  tags,
  from = 0,
  size = 20,
}) => {
  const mustClauses: any[] = [];
  const filterClauses: any[] = [];

  if (query) {
    mustClauses.push({
      multi_match: {
        query,
        fields: ['title^3', 'description', 'tags'],  // ^3 = 3x boost on title
        type: 'best_fields',
        fuzziness: 'AUTO',
      },
    });
  }

  if (category)   filterClauses.push({ term: { 'category.keyword': category } });
  if (tags?.length) filterClauses.push({ terms: { tags } });
  if (minPrice != null || maxPrice != null) {
    filterClauses.push({
      range: {
        price: {
          ...(minPrice != null ? { gte: minPrice } : {}),
          ...(maxPrice != null ? { lte: maxPrice } : {}),
        },
      },
    });
  }

  const response = await client.search({
    index: 'products',
    body: {
      from,
      size: Math.min(size, 100),
      query: {
        bool: {
          must:   mustClauses.length   ? mustClauses   : [{ match_all: {} }],
          filter: filterClauses.length ? filterClauses : [],
        },
      },
      aggs: {
        categories: { terms: { field: 'category.keyword', size: 20 } },
        price_stats: { stats: { field: 'price' } },
      },
      highlight: {
        fields: { title: {}, description: { number_of_fragments: 1, fragment_size: 150 } },
      },
      _source: ['id', 'title', 'category', 'price', 'tags'],  // Limit response fields
    },
  });

  return {
    hits: response.body.hits.hits.map((h: any) => ({
      ...h._source,
      highlight: h.highlight,
      score: h._score,
    })),
    totalHits: response.body.hits.total.value,
    aggregations: response.body.aggregations,
    from,
    size,
  };
});

OpenSearch limits from + size to 10,000 by default (controlled by index.max_result_window). MCP tools that need to paginate beyond 10,000 results must use the scroll API or point-in-time (PIT) with search_after.

Scroll API — export large result sets from MCP tool handlers

For MCP tools that need to export or process all documents matching a query (e.g., a data migration tool, a reporting tool), the scroll API provides a cursor-based mechanism that bypasses the max_result_window limit. Each scroll call returns a _scroll_id that must be passed to the next call. Always clear the scroll context when done to release memory on the OpenSearch cluster.

// MCP tool: export all matching documents via scroll
server.tool('export_products', async ({ category, batchCallback }) => {
  const allDocuments: any[] = [];
  let scrollId: string | undefined;

  try {
    // Initial search — opens the scroll context (2-minute window)
    const initialResponse = await client.search({
      index: 'products',
      scroll: '2m',  // Keep scroll context alive for 2 minutes between calls
      body: {
        size: 500,  // Documents per scroll batch
        query: {
          bool: {
            filter: category ? [{ term: { 'category.keyword': category } }] : [],
          },
        },
        _source: ['id', 'title', 'price', 'category'],
        sort: [{ _doc: 'asc' }],  // _doc sort is most efficient for scroll
      },
    });

    scrollId = initialResponse.body._scroll_id;
    let batch = initialResponse.body.hits.hits;

    while (batch.length > 0) {
      allDocuments.push(...batch.map((h: any) => h._source));

      // Fetch next batch
      const scrollResponse = await client.scroll({
        scroll_id: scrollId,
        scroll: '2m',  // Reset the 2-minute window on each call
      });

      scrollId = scrollResponse.body._scroll_id;
      batch = scrollResponse.body.hits.hits;
    }

  } finally {
    // Always clear scroll context — it holds memory on the cluster
    if (scrollId) {
      await client.clearScroll({ body: { scroll_id: scrollId } }).catch(() => {});
    }
  }

  return {
    total: allDocuments.length,
    documents: allDocuments,
  };
});

For paginated reads that span multiple MCP tool calls (where the scroll context would expire between calls), use point-in-time with search_after instead. Point-in-time provides a consistent snapshot without the time-bounded context of scroll.

Health probe — OpenSearch cluster health from an MCP server

OpenSearch's /_cluster/health endpoint returns the cluster's health status: green (all shards assigned and healthy), yellow (primary shards assigned, some replica shards unassigned — typical in single-node setups), or red (some primary shards unassigned — writes and reads may fail). For most MCP server deployments, yellow is acceptable and expected in single-node development environments.

async function checkOpenSearchHealth(): Promise<{
  healthy: boolean;
  status?: 'green' | 'yellow' | 'red';
  clusterName?: string;
  activeShards?: number;
  latencyMs?: number;
  error?: string;
}> {
  const start = Date.now();
  try {
    const response = await client.cluster.health({
      timeout: '5s',
      wait_for_status: 'yellow',  // Wait up to 5s for at least yellow status
    });

    const { status, cluster_name, active_shards } = response.body;
    return {
      healthy: status !== 'red',
      status,
      clusterName: cluster_name,
      activeShards: active_shards,
      latencyMs: Date.now() - start,
    };
  } catch (err: any) {
    return {
      healthy: false,
      error: err?.message ?? String(err),
    };
  }
}

async function startMcpServer() {
  const health = await checkOpenSearchHealth();
  if (!health.healthy) {
    console.error('OpenSearch unhealthy at startup:', health.status, health.error);
    process.exit(1);
  }
  console.log(
    \`OpenSearch ready (\${health.status} / \${health.activeShards} shards / \${health.latencyMs}ms). Registering MCP tools.\`
  );
}

server.tool('search_health', async () => {
  return checkOpenSearchHealth();
});

AliveMCP monitors your MCP server's OpenSearch-backed search tools by probing the cluster health endpoint every 60 seconds — catching index transitions to red status, shard allocation failures after node restarts, and network timeouts between the MCP server and the OpenSearch cluster before they cause agent tool call failures.