Kendra Custom Data Source API for MCP Servers — BatchPutDocument, Sync Jobs, and Programmatic Ingestion

When to use custom data source vs built-in connectors

Kendra provides over 40 built-in connectors. The decision between using one of them versus a CUSTOM data source comes down to two questions: does the connector exist, and do you need more control over what gets indexed than the connector allows? Built-in connectors handle authentication, pagination, and incremental crawls automatically — but they also make assumptions about document structure and ACL mapping that may not fit your use case.

A common hybrid pattern: use the built-in connector for the bulk of your content (e.g., the entire Confluence space on a nightly schedule) and a CUSTOM data source for time-sensitive supplemental content that must be searchable within minutes of creation (e.g., newly filed Jira incidents). Both data sources feed the same Kendra index and are searched together in a single QueryCommand call.

CDK: Creating a CUSTOM data source

Unlike connector-based data sources, a CUSTOM data source requires no S3 configuration, no schedule, and no crawl role on the Kendra side — because Kendra never reaches out to pull data. The only IAM role you need is on the caller side: the Lambda or ECS task that will push documents to Kendra.

import * as cdk from 'aws-cdk-lib';
import * as kendra from 'aws-cdk-lib/aws-kendra';
import * as iam from 'aws-cdk-lib/aws-iam';
import { Construct } from 'constructs';

export class KendraCustomDataSourceStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // Reference an existing Kendra index (created in a separate stack or imported)
    const indexId = cdk.Fn.importValue('KendraIndexId');

    // CUSTOM data source — no S3Configuration, no Schedule, no RoleArn required
    const customDs = new kendra.CfnDataSource(this, 'CustomDataSource', {
      indexId,
      name: 'custom-programmatic-source',
      type: 'CUSTOM',
      // No DataSourceConfiguration block needed for CUSTOM type
      // No Schedule — sync jobs are triggered programmatically via API
      description: 'Programmatic document ingestion via BatchPutDocument',
    });

    // IAM role for the Lambda / ECS task that will push documents
    // This is NOT attached to the data source — it's the caller's execution role
    const ingestionRole = new iam.Role(this, 'IngestionLambdaRole', {
      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
      managedPolicies: [
        iam.ManagedPolicy.fromAwsManagedPolicyName(
          'service-role/AWSLambdaBasicExecutionRole'
        ),
      ],
    });

    // Grant the ingestion Lambda all Kendra operations needed for custom data source sync
    ingestionRole.addToPolicy(new iam.PolicyStatement({
      effect: iam.Effect.ALLOW,
      actions: [
        'kendra:BatchPutDocument',
        'kendra:BatchDeleteDocument',
        'kendra:StartDataSourceSyncJob',
        'kendra:StopDataSourceSyncJob',
        'kendra:ListDataSourceSyncJobs',
        'kendra:DescribeDataSource',
      ],
      resources: [
        // Scope to the specific index and data source
        `arn:aws:kendra:${this.region}:${this.account}:index/${indexId}`,
        `arn:aws:kendra:${this.region}:${this.account}:index/${indexId}/data-source/*`,
      ],
    }));

    // Export the DataSource ID for use by the ingestion Lambda
    new cdk.CfnOutput(this, 'CustomDataSourceId', {
      value: customDs.attrId,
      exportName: 'KendraCustomDataSourceId',
      description: 'Kendra CUSTOM data source ID for BatchPutDocument ingestion',
    });

    new cdk.CfnOutput(this, 'IngestionRoleArn', {
      value: ingestionRole.roleArn,
      exportName: 'KendraIngestionLambdaRoleArn',
    });
  }
}

One important distinction: when you create a data source of type S3, SHAREPOINT, or any other built-in connector type, Kendra assumes a role to perform the crawl — that role needs S3 read access, secrets manager access for credentials, etc. With CUSTOM, Kendra never assumes a role and never reaches out. The RoleArn property on CfnDataSource is not applicable and should be omitted. The IAM permissions live entirely in your ingestion Lambda's execution role, scoped to the index and data source ARNs.

BatchPutDocument: document structure and attribute types

Each call to BatchPutDocumentCommand can submit up to 100 documents. The Documents array is the core payload — every element is a Kendra Document object. Getting the structure right matters: Kendra rejects malformed attribute values silently in FailedDocuments rather than throwing an exception, so it's worth validating locally before submission.

import {
  KendraClient,
  BatchPutDocumentCommand,
  DocumentAttributeValueType,
} from '@aws-sdk/client-kendra';
import crypto from 'crypto';

const kendra = new KendraClient({ region: process.env.AWS_REGION ?? 'us-east-1' });

const INDEX_ID = process.env.KENDRA_INDEX_ID!;
const DATA_SOURCE_ID = process.env.KENDRA_DATA_SOURCE_ID!;

// Build a stable document ID from a source URL or database primary key.
// IDs must be unique within the index; using a hash ensures stability across runs.
// Avoid characters: / # (causes INVALID_REQUEST); max length 2048 chars.
function stableDocId(sourceKey: string): string {
  // base64url encoding is safe for Kendra document IDs
  return crypto.createHash('sha256').update(sourceKey).digest('base64url');
}

interface SourceDocument {
  url: string;
  title: string;
  content: string;       // raw text or HTML
  contentType: 'PLAIN_TEXT' | 'HTML';
  category: string;
  tags: string[];
  priority: number;
  publishedAt: Date;
}

async function batchPutSourceDocuments(docs: SourceDocument[]) {
  if (docs.length === 0) return { indexed: 0, failed: [] };
  if (docs.length > 100) throw new Error('BatchPutDocument limit is 100 documents per call');

  const kendraDocuments = docs.map(doc => ({
    // Stable, unique string ID — never use a raw URL as-is (contains / and #)
    Id: stableDocId(doc.url),

    // Title shown in search results
    Title: doc.title,

    // Document content — raw bytes, max 2 MB
    Blob: Buffer.from(doc.content, 'utf-8'),

    // PLAIN_TEXT or HTML — Kendra parses HTML to extract text and strip tags
    ContentType: doc.contentType,

    // Custom attributes — Key must be defined in the index schema
    Attributes: [
      {
        Key: 'source_url',
        Value: { StringValue: doc.url },             // STRING
      },
      {
        Key: 'category',
        Value: { StringValue: doc.category },        // STRING
      },
      {
        Key: 'tags',
        Value: { StringListValue: doc.tags },        // STRING_LIST
      },
      {
        Key: 'priority_score',
        Value: { LongValue: BigInt(doc.priority) },  // LONG (must be BigInt in SDK v3)
      },
      {
        Key: '_last_updated_at',                     // built-in attribute for freshness boost
        Value: { DateValue: doc.publishedAt },       // DATE — must be a Date object, NOT a string
      },
    ],

    // Associate this document with our custom data source for _data_source_id filtering
    // This is set automatically by Kendra when you call BatchPutDocument within a sync job,
    // but explicit association ensures correct attribution
    ContentSourceDetails: {
      DataSourceId: DATA_SOURCE_ID,
    },
  }));

  const response = await kendra.send(new BatchPutDocumentCommand({
    IndexId: INDEX_ID,
    Documents: kendraDocuments,
  }));

  // FailedDocuments is a partial-failure list — Kendra processes the rest of the batch
  // even if some documents fail. Always inspect this array.
  const failedDocuments = response.FailedDocuments ?? [];

  for (const failed of failedDocuments) {
    console.error('BatchPutDocument partial failure', {
      documentId: failed.Id,
      errorCode: failed.ErrorCode,   // e.g., 'INVALID_REQUEST', 'ENDPOINT_DOCUMENT_SIZE_EXCEEDED'
      errorMessage: failed.ErrorMessage,
    });
  }

  return {
    indexed: docs.length - failedDocuments.length,
    failed: failedDocuments,
  };
}

Attribute value types in Kendra are strict at ingestion time. The four types map to SDK properties as follows:

Full sync job lifecycle with TypeScript

A sync job gives Kendra a logical boundary for a batch of document changes. The execution ID returned by StartDataSourceSyncJob is used internally by Kendra to track which documents were submitted in this run. After StopDataSourceSyncJob, Kendra finalizes the job and updates the index; you can then poll ListDataSourceSyncJobs until the status reaches SUCCEEDED or FAILED.

import {
  KendraClient,
  StartDataSourceSyncJobCommand,
  StopDataSourceSyncJobCommand,
  BatchPutDocumentCommand,
  ListDataSourceSyncJobsCommand,
  DataSourceSyncJobStatus,
} from '@aws-sdk/client-kendra';

const kendra = new KendraClient({ region: process.env.AWS_REGION ?? 'us-east-1' });
const INDEX_ID = process.env.KENDRA_INDEX_ID!;
const DATA_SOURCE_ID = process.env.KENDRA_DATA_SOURCE_ID!;

// Paginated source fetcher — replace with your actual data source
// (DynamoDB scan, REST API pages, database cursor, etc.)
async function* fetchSourcePages(): AsyncGenerator<SourceDocument[]> {
  let page = 1;
  while (true) {
    const batch = await fetchApiPage(page);  // your data source implementation
    if (batch.length === 0) break;
    yield batch;
    if (batch.length < 100) break;  // last page
    page++;
  }
}

interface SyncStats {
  executionId: string;
  totalIndexed: number;
  totalFailed: number;
  durationMs: number;
  finalStatus: string;
}

async function runFullSync(): Promise<SyncStats> {
  const startTime = Date.now();

  // Step 1: Start the sync job — get ExecutionId
  const startResp = await kendra.send(new StartDataSourceSyncJobCommand({
    Id: DATA_SOURCE_ID,
    IndexId: INDEX_ID,
  }));
  const executionId = startResp.ExecutionId!;
  console.log('Sync job started', { executionId });

  let totalIndexed = 0;
  let totalFailed = 0;
  const allFailedDocuments: Array<{ Id?: string; ErrorCode?: string; ErrorMessage?: string }> = [];

  try {
    // Step 2: Paginate through source data, push batches of 100
    let batchBuffer: SourceDocument[] = [];

    for await (const page of fetchSourcePages()) {
      batchBuffer.push(...page);

      // Flush complete batches of 100
      while (batchBuffer.length >= 100) {
        const batch = batchBuffer.splice(0, 100);
        const { indexed, failed } = await batchPutSourceDocuments(batch);
        totalIndexed += indexed;
        totalFailed += failed.length;
        allFailedDocuments.push(...failed);
        console.log('Batch pushed', { indexed, failed: failed.length, totalIndexed });
      }
    }

    // Flush the remaining partial batch
    if (batchBuffer.length > 0) {
      const { indexed, failed } = await batchPutSourceDocuments(batchBuffer);
      totalIndexed += indexed;
      totalFailed += failed.length;
      allFailedDocuments.push(...failed);
    }

  } finally {
    // Step 3: Always stop the sync job — even if an error occurred mid-sync.
    // Failure to stop leaves the data source locked (CONFLICT on next StartDataSourceSyncJob).
    await kendra.send(new StopDataSourceSyncJobCommand({
      Id: DATA_SOURCE_ID,
      IndexId: INDEX_ID,
    }));
    console.log('Sync job stopped', { executionId });
  }

  // Step 4: Poll until Kendra finalizes the job (SUCCEEDED or FAILED)
  const finalStatus = await waitForSyncJobCompletion(executionId);

  if (allFailedDocuments.length > 0) {
    console.warn('Some documents failed to index', allFailedDocuments);
  }

  return {
    executionId,
    totalIndexed,
    totalFailed,
    durationMs: Date.now() - startTime,
    finalStatus,
  };
}

async function waitForSyncJobCompletion(executionId: string): Promise<string> {
  const terminalStatuses: DataSourceSyncJobStatus[] = [
    DataSourceSyncJobStatus.SUCCEEDED,
    DataSourceSyncJobStatus.FAILED,
    DataSourceSyncJobStatus.INCOMPLETE,
    DataSourceSyncJobStatus.STOPPING,
  ];

  for (let attempt = 0; attempt < 60; attempt++) {
    await new Promise(r => setTimeout(r, 10_000));  // poll every 10 seconds

    const resp = await kendra.send(new ListDataSourceSyncJobsCommand({
      Id: DATA_SOURCE_ID,
      IndexId: INDEX_ID,
      MaxResults: 5,
    }));

    const job = resp.History?.find(h => h.ExecutionId === executionId);
    if (!job) continue;

    console.log('Sync job status', { executionId, status: job.Status, metrics: job.Metrics });

    if (terminalStatuses.includes(job.Status as DataSourceSyncJobStatus)) {
      return job.Status ?? 'UNKNOWN';
    }
  }

  throw new Error(`Sync job ${executionId} did not reach terminal status within 10 minutes`);
}

The Metrics field in the ListDataSourceSyncJobs response includes DocumentsAdded, DocumentsModified, DocumentsDeleted, and DocumentsFailed as string-encoded integers. These are authoritative counts from Kendra's side — your local totalIndexed counter tracks what you submitted, while Kendra's metrics reflect what was actually indexed. Discrepancies indicate that some documents were rejected after submission (e.g., during Kendra's internal processing phase).

Incremental updates without full resync

Running a full sync every time is expensive for large datasets. Kendra supports efficient incremental updates: calling BatchPutDocument for a document ID that already exists in the index updates it in place — Kendra deduplicates by Id and replaces the stored content and attributes. For deletions, use BatchDeleteDocumentCommand with a list of document IDs to remove. The pattern below maintains a last_synced_at watermark in DynamoDB and only pushes documents modified since the last run.

import {
  KendraClient,
  BatchDeleteDocumentCommand,
  QueryCommand,
  AttributeFilterOperator,
} from '@aws-sdk/client-kendra';
import {
  DynamoDBClient,
  GetItemCommand,
  PutItemCommand,
} from '@aws-sdk/client-dynamodb';

const ddb = new DynamoDBClient({ region: process.env.AWS_REGION });

const WATERMARK_TABLE = process.env.WATERMARK_TABLE!;
const WATERMARK_KEY = `kendra-sync-${DATA_SOURCE_ID}`;

async function getLastSyncedAt(): Promise<Date | null> {
  const resp = await ddb.send(new GetItemCommand({
    TableName: WATERMARK_TABLE,
    Key: { pk: { S: WATERMARK_KEY } },
  }));
  const ts = resp.Item?.last_synced_at?.S;
  return ts ? new Date(ts) : null;
}

async function setLastSyncedAt(ts: Date): Promise<void> {
  await ddb.send(new PutItemCommand({
    TableName: WATERMARK_TABLE,
    Item: {
      pk: { S: WATERMARK_KEY },
      last_synced_at: { S: ts.toISOString() },
    },
  }));
}

async function runIncrementalSync(): Promise<void> {
  const lastSyncedAt = await getLastSyncedAt();
  const syncStartedAt = new Date();

  // Fetch only documents modified after the watermark
  const modifiedDocs = await fetchModifiedSince(lastSyncedAt);
  console.log(`Incremental sync: ${modifiedDocs.length} modified documents`);

  if (modifiedDocs.length > 0) {
    // Start sync job, push only modified docs, stop
    await kendra.send(new StartDataSourceSyncJobCommand({ Id: DATA_SOURCE_ID, IndexId: INDEX_ID }));
    try {
      for (let i = 0; i < modifiedDocs.length; i += 100) {
        const batch = modifiedDocs.slice(i, i + 100);
        await batchPutSourceDocuments(batch);
      }
    } finally {
      await kendra.send(new StopDataSourceSyncJobCommand({ Id: DATA_SOURCE_ID, IndexId: INDEX_ID }));
    }
  }

  // Handle deletions: check for documents in Kendra that no longer exist in the source
  // Query Kendra for all documents from this data source, compare against current source IDs
  const deletedIds = await findDeletedDocumentIds(modifiedDocs);
  if (deletedIds.length > 0) {
    await kendra.send(new BatchDeleteDocumentCommand({
      IndexId: INDEX_ID,
      DocumentIdList: deletedIds,
      DataSourceSyncJobMetricTarget: {
        DataSourceId: DATA_SOURCE_ID,
        DataSourceSyncJobId: 'incremental-delete',
      },
    }));
    console.log(`Deleted ${deletedIds.length} documents from Kendra`);
  }

  // Update the watermark only after a successful sync
  await setLastSyncedAt(syncStartedAt);
}

// Use Kendra QueryCommand with AttributeFilter on _data_source_id to enumerate
// documents from this data source and find IDs no longer present in the source
async function findDeletedDocumentIds(currentDocs: SourceDocument[]): Promise<string[]> {
  const currentIds = new Set(currentDocs.map(d => stableDocId(d.url)));

  const kendraIds = new Set<string>();
  let pageNumber = 1;

  while (true) {
    const resp = await kendra.send(new QueryCommand({
      IndexId: INDEX_ID,
      QueryText: '*',    // match-all query to enumerate documents
      PageSize: 100,
      PageNumber: pageNumber,
      AttributeFilter: {
        EqualsTo: {
          Key: '_data_source_id',
          Value: { StringValue: DATA_SOURCE_ID },
        },
      },
    }));

    for (const item of resp.ResultItems ?? []) {
      if (item.Id) kendraIds.add(item.Id);
    }

    if (!resp.ResultItems?.length || pageNumber >= 10) break;  // Kendra caps at page 10
    pageNumber++;
  }

  // Any ID in Kendra but not in current source has been deleted
  return [...kendraIds].filter(id => !currentIds.has(id));
}

The watermark approach is robust to Lambda timeouts and retries: if the sync crashes mid-run, the watermark is not updated, so the next invocation re-processes the same window. This can produce duplicate BatchPutDocument calls for documents that were already indexed in the failed run — that is safe because Kendra deduplicates by ID. The only scenario that requires care is if the source document was deleted between the failed and retry runs; handle this by running the deletion check on every sync, not just after successful completion.

Common failures