Bedrock Batch Inference in MCP Servers — CreateModelInvocationJob, S3 JSONL, async results

When to use batch inference vs on-demand streaming

Bedrock batch inference is optimized for throughput, not latency. It is the right choice when:

• Request count is large (100+) and latency is not the constraint. Batch jobs process records concurrently up to 100 at a time, but total job turnaround is minutes to hours depending on queue depth and corpus size.
• The MCP tool is a background analysis pipeline. An MCP tool that generates a "weekly health summary" for all monitored endpoints, classifies server error messages, or scores uptime quality across a registry is a natural batch job — the user doesn't need a result in 30 seconds.
• Cost is a primary concern. At 50% of on-demand pricing, batch inference halves the token cost for supported models. A corpus of 10,000 requests at $0.003/1k input tokens saves $15 vs on-demand — significant at scale.

On-demand streaming is the right choice when the MCP tool must return results within a single tool-call timeout (typically 30–120 seconds), or when requests are user-interactive. Do not use batch inference when you need results in the same session that submitted the job.

S3 input JSONL format

The input file is a JSONL (newline-delimited JSON) file uploaded to an S3 bucket. Each line is a self-contained JSON object with two required fields: recordId (a unique string identifier for this record) and modelInput (the raw model request body, exactly as you would send to InvokeModel for that model family).

// Generate a JSONL batch input file for Claude via Bedrock batch inference
// modelInput uses the Anthropic Messages API format (same as InvokeModel)

interface BatchRecord {
  recordId: string;
  modelInput: {
    anthropic_version: string;
    max_tokens: number;
    messages: Array<{ role: 'user' | 'assistant'; content: string }>;
    system?: string;
  };
}

function buildBatchInputLine(
  recordId: string,
  prompt: string,
  systemPrompt?: string,
): string {
  const record: BatchRecord = {
    recordId,
    modelInput: {
      anthropic_version: 'bedrock-2023-05-31',
      max_tokens: 1024,
      messages: [{ role: 'user', content: prompt }],
      ...(systemPrompt ? { system: systemPrompt } : {}),
    },
  };
  return JSON.stringify(record);
}

// Example: generate summaries for a list of MCP server error logs
function buildErrorSummaryBatch(
  errorLogs: Array<{ serverId: string; log: string }>,
): string {
  const systemPrompt = 'You are an expert at analyzing MCP server error logs. '
    + 'Summarize the root cause in one sentence and suggest a fix.';

  return errorLogs
    .map(({ serverId, log }) =>
      buildBatchInputLine(serverId, `Error log:\n${log}`, systemPrompt)
    )
    .join('\n');
}

// Upload to S3
import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3';

const s3 = new S3Client({ region: 'us-east-1' });

async function uploadBatchInput(
  bucket: string,
  key: string,
  jsonlContent: string,
): Promise {
  await s3.send(new PutObjectCommand({
    Bucket: bucket,
    Key: key,
    Body: jsonlContent,
    ContentType: 'application/jsonlines',
  }));
  console.log(`Uploaded batch input: s3://${bucket}/${key}`);
}

Important constraints: each record's modelInput must be valid JSON matching the model-specific request format. For Claude models via Bedrock, this is the Anthropic Messages API format (not the Converse API format). For Amazon Titan, it is the Titan-specific body. The S3 file must be readable by the IAM role you pass to the batch job — same account S3 access is typical, but cross-account is also supported with bucket policies.

IAM role and CreateModelInvocationJob

Bedrock batch inference requires a dedicated IAM role that Bedrock can assume via a trust relationship. This role needs bedrock:InvokeModel on the target model plus s3:GetObject on the input bucket and s3:PutObject on the output bucket.

// IAM trust policy for the Bedrock batch inference role
// (Set up in AWS Console or CDK — this is the trust policy JSON)
const trustPolicy = {
  Version: '2012-10-17',
  Statement: [{
    Effect: 'Allow',
    Principal: { Service: 'bedrock.amazonaws.com' },
    Action: 'sts:AssumeRole',
    Condition: {
      StringEquals: { 'aws:SourceAccount': ACCOUNT_ID },
      ArnLike: { 'aws:SourceArn': `arn:aws:bedrock:us-east-1:${ACCOUNT_ID}:*` },
    },
  }],
};

// Permission policy attached to the role
const permissionPolicy = {
  Version: '2012-10-17',
  Statement: [
    {
      Sid: 'BedrockInvoke',
      Effect: 'Allow',
      Action: ['bedrock:InvokeModel'],
      Resource: [
        'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-sonnet-20241022-v2:0',
        'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-haiku-20240307-v1:0',
      ],
    },
    {
      Sid: 'S3Input',
      Effect: 'Allow',
      Action: ['s3:GetObject', 's3:ListBucket'],
      Resource: [
        `arn:aws:s3:::${BATCH_INPUT_BUCKET}`,
        `arn:aws:s3:::${BATCH_INPUT_BUCKET}/*`,
      ],
    },
    {
      Sid: 'S3Output',
      Effect: 'Allow',
      Action: ['s3:PutObject'],
      Resource: [`arn:aws:s3:::${BATCH_OUTPUT_BUCKET}/*`],
    },
  ],
};

// Create the batch job
import {
  BedrockClient,
  CreateModelInvocationJobCommand,
  GetModelInvocationJobCommand,
} from '@aws-sdk/client-bedrock';

const bedrock = new BedrockClient({ region: 'us-east-1' });

async function submitBatchJob(
  inputS3Uri: string,   // s3://bucket/path/input.jsonl
  outputS3Uri: string,  // s3://bucket/path/output/ (prefix, not file)
  jobName: string,
): Promise {
  const response = await bedrock.send(new CreateModelInvocationJobCommand({
    modelId: 'anthropic.claude-3-5-sonnet-20241022-v2:0',
    jobName,
    roleArn: process.env.BEDROCK_BATCH_ROLE_ARN!,
    inputDataConfig: {
      s3InputDataConfig: {
        s3Uri: inputS3Uri,
        s3InputFormat: 'JSONLines',
      },
    },
    outputDataConfig: {
      s3OutputDataConfig: {
        s3Uri: outputS3Uri,
      },
    },
  }));

  const jobArn = response.jobArn;
  if (!jobArn) throw new Error('No jobArn in CreateModelInvocationJob response');
  console.log(`Batch job created: ${jobArn}`);
  return jobArn;
}

The jobName must be unique within the account. A common pattern is to use a timestamp plus a hash of the input content: mcp-analysis-${Date.now()}-${hash}. Job ARNs are the durable identifier for the job — store the ARN returned by CreateModelInvocationJobCommand to poll status and retrieve results later.

Job polling and EventBridge completion notification

For an MCP tool that submits a batch job and needs to return a job handle to the user, the tool should return the job ARN immediately and expose a separate polling tool. For background pipelines that trigger downstream actions on completion, EventBridge is a better pattern than polling.

// Poll job status — use this in a background worker, not in a synchronous MCP tool handler
type JobStatus = 'Submitted' | 'InProgress' | 'Completed' | 'Failed' | 'Stopped';

async function pollJobUntilComplete(
  jobArn: string,
  pollIntervalMs = 30_000,
  timeoutMs = 6 * 60 * 60 * 1000, // 6 hours max
): Promise {
  const start = Date.now();

  while (Date.now() - start < timeoutMs) {
    const response = await bedrock.send(new GetModelInvocationJobCommand({
      jobIdentifier: jobArn,
    }));

    const status = response.status as JobStatus;
    console.log(`Job ${jobArn}: ${status}`);

    if (status === 'Completed' || status === 'Failed' || status === 'Stopped') {
      return status;
    }

    await new Promise(r => setTimeout(r, pollIntervalMs));
  }

  throw new Error(`Job ${jobArn} did not complete within timeout`);
}

// EventBridge rule for push notification (set up via CDK or Console)
// Event pattern:
// {
//   "source": ["aws.bedrock"],
//   "detail-type": ["Batch Inference Job State Change"],
//   "detail": {
//     "status": ["Completed", "Failed"]
//   }
// }
// Target: Lambda function that reads output S3 and processes results

// MCP tool: submit job and return ARN for later retrieval
async function submitAnalysisJob(
  documents: Array<{ id: string; content: string }>,
): Promise<{ jobArn: string; inputKey: string }> {
  const batchContent = buildErrorSummaryBatch(
    documents.map(d => ({ serverId: d.id, log: d.content }))
  );

  const inputKey = `batch-inputs/${Date.now()}.jsonl`;
  await uploadBatchInput(BATCH_BUCKET, inputKey, batchContent);

  const jobArn = await submitBatchJob(
    `s3://${BATCH_BUCKET}/${inputKey}`,
    `s3://${BATCH_BUCKET}/batch-outputs/${Date.now()}/`,
    `mcp-analysis-${Date.now()}`,
  );

  return { jobArn, inputKey };
}

Processing output results and error records

When the job completes, Bedrock writes a JSONL output file to the S3 output prefix. Each line corresponds to an input record and contains either a successful model output or an error. Both successes and errors use the same JSON structure.

import { GetObjectCommand } from '@aws-sdk/client-s3';

interface BatchOutputRecord {
  recordId: string;
  modelOutput?: {
    id: string;
    type: 'message';
    role: 'assistant';
    content: Array<{ type: 'text'; text: string }>;
    model: string;
    stop_reason: string;
    usage: { input_tokens: number; output_tokens: number };
  };
  error?: {
    errorCode: string;
    errorMessage: string;
  };
}

async function processJobOutput(
  outputS3Uri: string,
): Promise<{
  successes: Array<{ recordId: string; text: string }>;
  failures: Array<{ recordId: string; errorCode: string; errorMessage: string }>;
}> {
  // Parse s3://bucket/prefix/ → list objects
  const [bucket, prefix] = outputS3Uri.replace('s3://', '').split(/\/(.*)/);

  // List output objects (Bedrock writes one or more JSONL files per job)
  const { ListObjectsV2Command } = await import('@aws-sdk/client-s3');
  const { Contents } = await s3.send(new ListObjectsV2Command({ Bucket: bucket, Prefix: prefix }));

  const successes: Array<{ recordId: string; text: string }> = [];
  const failures: Array<{ recordId: string; errorCode: string; errorMessage: string }> = [];

  for (const obj of Contents ?? []) {
    if (!obj.Key?.endsWith('.jsonl')) continue;

    const { Body } = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: obj.Key }));
    const text = await Body?.transformToString('utf-8') ?? '';

    for (const line of text.split('\n').filter(Boolean)) {
      const record: BatchOutputRecord = JSON.parse(line);

      if (record.error) {
        failures.push({
          recordId: record.recordId,
          errorCode: record.error.errorCode,
          errorMessage: record.error.errorMessage,
        });
      } else if (record.modelOutput?.content?.[0]?.text) {
        successes.push({
          recordId: record.recordId,
          text: record.modelOutput.content[0].text,
        });
      }
    }
  }

  console.log(`Processed: ${successes.length} successes, ${failures.length} failures`);
  return { successes, failures };
}

Common error codes in the output: ThrottledException (record failed due to throttling — retry it via on-demand), ModelTimeoutException (model took too long for this record — reduce max_tokens or simplify the prompt), ValidationException (malformed modelInput — fix the input record). Failed records do not cause the entire job to fail — the job status is still Completed even if some records errored. Check the failure count before declaring success.

Supported models and job limits

Key limits: 50,000 records per job; 1 concurrent running job per account per region (additional jobs queue); job expires after 7 days if not completed (failed records cause quota consumption but the job can complete with partial results). For production pipelines, set up CloudWatch Alarms on the EventBridge job-state-change events and alert on Failed jobs.