bulk

Efficiently execute multiple Elasticsearch document operations—create, update, delete—in one API call, reducing requests and optimizing performance.

Instructions

Perform multiple document operations (create, update, delete) in a single API call

Input Schema

TableJSON Schema

Name	Required	Description	Default
`operations`	Yes	Array of operations to perform in bulk
`pipeline`	No	Optional pipeline to use for preprocessing documents

Implementation Reference

src/index.ts:808-933 (handler)

Main handler logic for the 'bulk' tool: validates input operations, builds Elasticsearch bulk operations array in correct format, calls esService.bulk, processes the response to summarize successes/failures, formats and returns text output.

async ({ operations, pipeline }) => {
  try {
    // Validate operations
    operations.forEach((op, idx) => {
      if ((op.action === "update" || op.action === "delete") && !op.id) {
        throw new Error(
          `Operation #${idx + 1} (${op.action}): Document ID is required`
        );
      }
      if (
        (op.action === "index" ||
          op.action === "create" ||
          op.action === "update") &&
        !op.document
      ) {
        throw new Error(
          `Operation #${idx + 1} (${op.action}): Document body is required`
        );
      }
    });

    // Build the bulk operations array
    const bulkOperations: any[] = [];

    operations.forEach((op) => {
      const actionMeta: any = { _index: op.index };
      if (op.id) actionMeta._id = op.id;

      bulkOperations.push({ [op.action]: actionMeta });

      if (op.action !== "delete") {
        if (op.action === "update") {
          bulkOperations.push({ doc: op.document });
        } else {
          bulkOperations.push(op.document);
        }
      }
    });

    // Execute the bulk operation
    const response = await esService.bulk(bulkOperations, pipeline);

    // Process the response
    const summary = {
      took: response.took,
      errors: response.errors,
      successes: 0,
      failures: 0,
      actionResults: [] as any[],
    };

    // Count successes and failures
    response.items.forEach((item: any, idx: number) => {
      const actionType = Object.keys(item)[0];
      const result = item[actionType as keyof typeof item] as any;

      if (!result) return;

      if (result.error) {
        summary.failures++;
        summary.actionResults.push({
          operation: idx,
          action: actionType,
          id: result._id || "unknown",
          index: result._index || "unknown",
          status: result.status || 0,
          error: {
            type: result.error?.type || "unknown_error",
            reason: result.error?.reason || "Unknown error",
          },
        });
      } else {
        summary.successes++;
        summary.actionResults.push({
          operation: idx,
          action: actionType,
          id: result._id || "unknown",
          index: result._index || "unknown",
          status: result.status || 0,
          result: result.result || "unknown",
        });
      }
    });

    // Format the response
    let resultText = `Bulk operation completed in ${summary.took}ms\n`;
    resultText += `- Total operations: ${operations.length}\n`;
    resultText += `- Successful: ${summary.successes}\n`;
    resultText += `- Failed: ${summary.failures}\n`;

    // Add failure details
    if (summary.failures > 0) {
      resultText += "\nFailures:\n";
      const failures = summary.actionResults.filter((r) => r.error);
      failures.slice(0, 5).forEach((failure, idx) => {
        resultText += `${idx + 1}. Operation #${failure.operation} (${
          failure.action
        }): ${failure.error.reason} [${failure.error.type}]\n`;
      });

      if (failures.length > 5) {
        resultText += `...and ${failures.length - 5} more failures.\n`;
      }
    }

    return {
      content: [{ type: "text", text: resultText }],
    };
  } catch (error) {
    console.error(
      `Bulk operation failed: ${
        error instanceof Error ? error.message : String(error)
      }`
    );
    return {
      content: [
        {
          type: "text",
          text: `Error: ${
            error instanceof Error ? error.message : String(error)
          }`,
        },
      ],
    };
  }
}

src/index.ts:773-807 (schema)

Input schema using Zod for the 'bulk' tool parameters: operations array and optional pipeline.

{
  operations: z
    .array(
      z.object({
        action: z
          .enum(["index", "create", "update", "delete"])
          .describe(
            "The action to perform: index (create/replace), create (fail if exists), update, or delete"
          ),
        index: z
          .string()
          .trim()
          .min(1, "Index name is required")
          .describe("Name of the Elasticsearch index for this operation"),
        id: z
          .string()
          .optional()
          .describe(
            "Document ID (required for update and delete, optional for index/create)"
          ),
        document: z
          .record(z.any())
          .optional()
          .describe(
            "Document body (required for index/create/update, not used for delete)"
          ),
      })
    )
    .min(1, "At least one operation is required")
    .describe("Array of operations to perform in bulk"),
  pipeline: z
    .string()
    .optional()
    .describe("Optional pipeline to use for preprocessing documents"),
},

src/index.ts:770-934 (registration)

Registration of the 'bulk' MCP tool via server.tool, specifying name, description, input schema, and handler function.

server.tool(
  "bulk",
  "Perform multiple document operations (create, update, delete) in a single API call",
  {
    operations: z
      .array(
        z.object({
          action: z
            .enum(["index", "create", "update", "delete"])
            .describe(
              "The action to perform: index (create/replace), create (fail if exists), update, or delete"
            ),
          index: z
            .string()
            .trim()
            .min(1, "Index name is required")
            .describe("Name of the Elasticsearch index for this operation"),
          id: z
            .string()
            .optional()
            .describe(
              "Document ID (required for update and delete, optional for index/create)"
            ),
          document: z
            .record(z.any())
            .optional()
            .describe(
              "Document body (required for index/create/update, not used for delete)"
            ),
        })
      )
      .min(1, "At least one operation is required")
      .describe("Array of operations to perform in bulk"),
    pipeline: z
      .string()
      .optional()
      .describe("Optional pipeline to use for preprocessing documents"),
  },
  async ({ operations, pipeline }) => {
    try {
      // Validate operations
      operations.forEach((op, idx) => {
        if ((op.action === "update" || op.action === "delete") && !op.id) {
          throw new Error(
            `Operation #${idx + 1} (${op.action}): Document ID is required`
          );
        }
        if (
          (op.action === "index" ||
            op.action === "create" ||
            op.action === "update") &&
          !op.document
        ) {
          throw new Error(
            `Operation #${idx + 1} (${op.action}): Document body is required`
          );
        }
      });

      // Build the bulk operations array
      const bulkOperations: any[] = [];

      operations.forEach((op) => {
        const actionMeta: any = { _index: op.index };
        if (op.id) actionMeta._id = op.id;

        bulkOperations.push({ [op.action]: actionMeta });

        if (op.action !== "delete") {
          if (op.action === "update") {
            bulkOperations.push({ doc: op.document });
          } else {
            bulkOperations.push(op.document);
          }
        }
      });

      // Execute the bulk operation
      const response = await esService.bulk(bulkOperations, pipeline);

      // Process the response
      const summary = {
        took: response.took,
        errors: response.errors,
        successes: 0,
        failures: 0,
        actionResults: [] as any[],
      };

      // Count successes and failures
      response.items.forEach((item: any, idx: number) => {
        const actionType = Object.keys(item)[0];
        const result = item[actionType as keyof typeof item] as any;

        if (!result) return;

        if (result.error) {
          summary.failures++;
          summary.actionResults.push({
            operation: idx,
            action: actionType,
            id: result._id || "unknown",
            index: result._index || "unknown",
            status: result.status || 0,
            error: {
              type: result.error?.type || "unknown_error",
              reason: result.error?.reason || "Unknown error",
            },
          });
        } else {
          summary.successes++;
          summary.actionResults.push({
            operation: idx,
            action: actionType,
            id: result._id || "unknown",
            index: result._index || "unknown",
            status: result.status || 0,
            result: result.result || "unknown",
          });
        }
      });

      // Format the response
      let resultText = `Bulk operation completed in ${summary.took}ms\n`;
      resultText += `- Total operations: ${operations.length}\n`;
      resultText += `- Successful: ${summary.successes}\n`;
      resultText += `- Failed: ${summary.failures}\n`;

      // Add failure details
      if (summary.failures > 0) {
        resultText += "\nFailures:\n";
        const failures = summary.actionResults.filter((r) => r.error);
        failures.slice(0, 5).forEach((failure, idx) => {
          resultText += `${idx + 1}. Operation #${failure.operation} (${
            failure.action
          }): ${failure.error.reason} [${failure.error.type}]\n`;
        });

        if (failures.length > 5) {
          resultText += `...and ${failures.length - 5} more failures.\n`;
        }
      }

      return {
        content: [{ type: "text", text: resultText }],
      };
    } catch (error) {
      console.error(
        `Bulk operation failed: ${
          error instanceof Error ? error.message : String(error)
        }`
      );
      return {
        content: [
          {
            type: "text",
            text: `Error: ${
              error instanceof Error ? error.message : String(error)
            }`,
          },
        ],
      };
    }
  }
);

src/utils/elasticsearchService.ts:228-234 (helper)
Helper method in ElasticsearchService that executes the Elasticsearch bulk API call with automatic refresh and optional pipeline.
```
async bulk(operations: any[], pipeline?: string): Promise<any> {
  return await this.client.bulk({
    refresh: true,
    pipeline,
    operations,
  });
}
```
src/types.ts:44-48 (schema)
TypeScript interface defining the structure of the bulk operation result from Elasticsearch.
```
export interface BulkOperationResult {
  took: number;
  errors: boolean;
  items: Array<Record<string, any>>;
}
```

octodet-elasticsearch-mcp

bulk

Instructions

Input Schema

Implementation Reference

Other Tools

Related Tools

Latest Blog Posts

MCP directory API