MCP Server Anthropic Batch API — async bulk processing, custom_id, result streaming

Creating a batch and returning the batch ID

The batch create call is synchronous and returns within a few seconds. It accepts up to 10,000 requests in one call. The correct MCP pattern is a two-tool design: one tool submits the batch and returns the ID, a second tool polls for completion and retrieves results. This avoids holding the tool handler open for the batch duration.

import Anthropic from '@anthropic-ai/sdk';
import { z } from 'zod';
import { randomUUID } from 'crypto';

const anthropic = new Anthropic();

// Tool 1: submit items for batch analysis, return a batch_id immediately
server.tool(
  'batch_classify_texts',
  {
    texts: z.array(z.object({
      id:   z.string(),     // caller-provided stable ID — becomes custom_id
      text: z.string().max(10_000),
    })).min(1).max(10_000),
    category_list: z.array(z.string()).min(2).max(20),
  },
  async ({ texts, category_list }) => {
    const categories = category_list.join(', ');
    const system = `Classify the provided text into exactly one of: ${categories}.
Respond with a JSON object: {"category": "...", "confidence": 0-1}`;

    // Build one request per input item
    const requests = texts.map(({ id, text }) => ({
      custom_id: id,   // must be unique within the batch; used to correlate results
      params: {
        model:      'claude-haiku-4-5-20251001',  // Haiku: cheapest per token
        max_tokens: 64,
        system,
        messages: [{ role: 'user', content: text }],
      },
    }));

    const batch = await anthropic.messages.batches.create({ requests });

    // Batch is now queued. Return immediately — do NOT poll here.
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          batch_id:          batch.id,
          status:            batch.processing_status,  // 'in_progress'
          request_count:     batch.request_counts.processing,
          expires_at:        batch.expires_at,
          check_tool:        'get_batch_results',
          message:           'Submit batch_id to get_batch_results when ready.',
        }),
      }],
    };
  }
);

// Tool 2: poll status, stream results when complete
server.tool(
  'get_batch_results',
  { batch_id: z.string() },
  async ({ batch_id }) => {
    const batch = await anthropic.messages.batches.retrieve(batch_id);

    if (batch.processing_status !== 'ended') {
      // Return progress — agent should retry later
      const counts = batch.request_counts;
      return {
        content: [{
          type: 'text',
          text: JSON.stringify({
            status:     batch.processing_status,
            processing: counts.processing,
            succeeded:  counts.succeeded,
            errored:    counts.errored,
            message:    `Batch still running. Retry in 60 seconds.`,
          }),
        }],
      };
    }

    // Batch is done — stream results
    const results: Array<{ custom_id: string; category?: string; confidence?: number; error?: string }> = [];

    for await (const result of anthropic.messages.batches.results(batch_id)) {
      if (result.result.type === 'succeeded') {
        const text = result.result.message.content.find(b => b.type === 'text')?.text ?? '{}';
        try {
          const parsed = JSON.parse(text);
          results.push({ custom_id: result.custom_id, ...parsed });
        } catch {
          results.push({ custom_id: result.custom_id, error: 'parse_failed', raw: text });
        }
      } else {
        // result.result.type === 'errored' | 'expired' | 'canceled'
        results.push({ custom_id: result.custom_id, error: result.result.type });
      }
    }

    return { content: [{ type: 'text', text: JSON.stringify(results) }] };
  }
);

The async iterator from anthropic.messages.batches.results() streams JSONL lines from the Anthropic API. Each line is one { custom_id, result } object. Results are not guaranteed to arrive in the same order as requests — always correlate via custom_id, never by position.

Deduplication and custom_id constraints

The Batch API does not deduplicate requests with the same custom_id — it processes every item in the array, even if two share the same ID. The API validates that all custom_id values within a single batch are unique and returns a 400 error if they are not. But if you submit two batches with the same item under the same custom_id, both process independently.

import { randomUUID } from 'crypto';

// Safe approach: derive custom_id deterministically from content hash,
// so re-submitting the same item with the same ID is idempotent at YOUR layer
// (the API still processes it, but your dedup logic can skip inserting the result twice).
import { createHash } from 'crypto';

function stableId(text: string): string {
  return createHash('sha256').update(text).digest('hex').slice(0, 64);
}

const requests = texts.map(({ text }) => ({
  custom_id: stableId(text),  // deterministic; same text = same ID
  params: { model: 'claude-haiku-4-5-20251001', max_tokens: 64, messages: [{ role: 'user', content: text }] },
}));

// Deduplicate before creating the batch — the API rejects duplicate custom_ids
const seen = new Set<string>();
const deduped = requests.filter(r => {
  if (seen.has(r.custom_id)) return false;
  seen.add(r.custom_id);
  return true;
});

Using content-derived IDs also lets you build a simple result cache: if a custom_id already has a stored result, skip re-submitting it and return the cached answer directly. This is particularly useful for classification tasks where the same text appears in multiple batches over time.

Cancellation and batch expiry

Batches that are still in progress can be cancelled with anthropic.messages.batches.cancel(batchId). After cancellation, processing_status moves to canceling and then ended — results for already-processed items are still available, cancelled items appear with result.type === 'canceled'. Batches expire 24 hours after creation regardless of status.

// Cancel a running batch — e.g. when the user changes their mind mid-session
server.tool(
  'cancel_batch',
  { batch_id: z.string() },
  async ({ batch_id }) => {
    const batch = await anthropic.messages.batches.cancel(batch_id);
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          batch_id,
          status:    batch.processing_status,  // 'canceling' immediately, 'ended' shortly after
          counts:    batch.request_counts,
          message:   'Batch is being cancelled. Completed items are still retrievable.',
        }),
      }],
    };
  }
);

// List all recent batches (useful for showing the agent active work)
server.tool(
  'list_batches',
  { limit: z.number().int().min(1).max(100).default(20) },
  async ({ limit }) => {
    const page = await anthropic.messages.batches.list({ limit });
    const summary = page.data.map(b => ({
      id:      b.id,
      status:  b.processing_status,
      counts:  b.request_counts,
      created: b.created_at,
      expires: b.expires_at,
    }));
    return { content: [{ type: 'text', text: JSON.stringify(summary) }] };
  }
);

When a batch expires (after 24 hours), its results are no longer retrievable from the API. If your MCP workflow produces batch IDs that agents might query hours later, persist the results to your own storage (SQLite, S3) as soon as the batch reaches ended, and serve from there instead of re-fetching from Anthropic.

Cost model and when to use batching

Batch API requests cost 50% less than equivalent synchronous Messages API calls. The trade-off is latency: batch jobs can take anywhere from 1 minute (small batches, low API load) to several hours (large batches, peak usage times). The Batch API is not suitable for interactive agent workflows where the user is waiting for a response, but it is ideal for MCP tools that perform bulk analysis as a background job.