Kendra S3 Data Source Connector — Document Sync, Metadata Files, and ACL Mapping for MCP Servers

S3 connector setup with CDK

The CfnDataSource construct wires the S3 connector to the Kendra index. It requires an IAM role that Kendra can assume to read from the S3 bucket — the trust policy must name kendra.amazonaws.com as the principal, and the role must have s3:GetObject and s3:ListBucket on the specific bucket. Using wildcards in the bucket ARN is not recommended: Kendra only needs access to the configured bucket, and a scoped policy makes the data source's access footprint auditable.

The S3DataSourceConfiguration object has three configuration sections: BucketConfiguration (the bucket name and optional inclusion/exclusion patterns), DocumentsMetadataConfiguration (the S3 prefix where metadata sidecar files live), and AccessControlListConfiguration (the S3 key path of the ACL mapping file for group-to-document mappings). All three sections are optional — a minimal connector only requires the bucket name — but metadata and ACL configurations are almost always needed in production.

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 * as s3 from 'aws-cdk-lib/aws-s3';
import { Construct } from 'constructs';

interface KendraS3ConnectorProps {
  indexId: string;
  documentsBucket: s3.IBucket;
  // S3 prefix where documents live, e.g. 'docs/'
  documentsPrefix: string;
  // S3 prefix where metadata sidecars live, e.g. 'docs-metadata/'
  metadataPrefix: string;
}

export class KendraS3Connector extends Construct {
  public readonly dataSourceId: string;

  constructor(scope: Construct, id: string, props: KendraS3ConnectorProps) {
    super(scope, id);

    const stack = cdk.Stack.of(this);

    // IAM role that Kendra assumes to read from S3
    const dataSourceRole = new iam.Role(this, 'KendraS3DataSourceRole', {
      assumedBy: new iam.ServicePrincipal('kendra.amazonaws.com'),
      description: 'Role assumed by Amazon Kendra to read documents from S3',
    });

    // Allow Kendra to list and read objects from the documents bucket
    props.documentsBucket.grantRead(dataSourceRole);

    // Kendra also needs to list the bucket root to discover objects
    dataSourceRole.addToPolicy(new iam.PolicyStatement({
      sid: 'KendraS3BucketList',
      effect: iam.Effect.ALLOW,
      actions: ['s3:ListBucket'],
      resources: [props.documentsBucket.bucketArn],
    }));

    // Allow Kendra to call kendra:BatchPutDocument and kendra:BatchDeleteDocument
    // on the index (required for the data source sync to write back to the index)
    dataSourceRole.addToPolicy(new iam.PolicyStatement({
      sid: 'KendraIndexWrite',
      effect: iam.Effect.ALLOW,
      actions: [
        'kendra:BatchPutDocument',
        'kendra:BatchDeleteDocument',
      ],
      resources: [
        `arn:aws:kendra:${stack.region}:${stack.account}:index/${props.indexId}`,
      ],
    }));

    // The S3 data source
    const cfnDataSource = new kendra.CfnDataSource(this, 'S3DataSource', {
      indexId: props.indexId,  // token reference — registers CDK dependency on the index
      name: 'mcp-s3-documents',
      type: 'S3',
      roleArn: dataSourceRole.roleArn,
      description: 'S3 document corpus for MCP enterprise search tool',

      dataSourceConfiguration: {
        s3Configuration: {
          bucketName: props.documentsBucket.bucketName,

          // Restrict indexing to a specific prefix
          inclusionPrefixes: [props.documentsPrefix],

          // Optionally exclude patterns — supports glob-style wildcards
          // exclusionPatterns: ['*.tmp', '*.draft', 'archive/*'],

          // Metadata sidecar configuration
          // Sidecars must live in this prefix, mirroring the document prefix structure.
          // IMPORTANT: metadataPrefix MUST end with '/'
          documentsMetadataConfiguration: {
            s3Prefix: props.metadataPrefix,
          },

          // ACL configuration — path to the access control list JSON file
          // This file maps group names to document sets; see ACL section below.
          // accessControlListConfiguration: {
          //   keyPath: 'acl/kendra-acl-mapping.json',
          // },
        },
      },

      // Sync schedule options:
      // 'cron(0 * * * ? *)'  — every hour at minute 0
      // 'cron(0 6 * * ? *)'  — daily at 06:00 UTC
      // 'ON_DEMAND'           — no automatic schedule; trigger via API
      schedule: 'cron(0 2 * * ? *)',  // daily at 02:00 UTC
    });

    this.dataSourceId = cfnDataSource.attrId;

    new cdk.CfnOutput(this, 'DataSourceId', {
      value: cfnDataSource.attrId,
      description: 'Kendra S3 data source ID',
    });
  }
}

The inclusionPrefixes and exclusionPatterns arrays give you fine-grained control over which S3 objects Kendra indexes. Inclusion prefixes are path prefixes (not glob patterns) — docs/ matches all objects under docs/. Exclusion patterns support glob wildcards: *.tmp excludes all .tmp files anywhere in the bucket, and archive/* excludes the entire archive/ prefix. Exclusion patterns are evaluated after inclusion — a file that matches both an inclusion prefix and an exclusion pattern is excluded. If neither inclusion nor exclusion is configured, Kendra crawls the entire bucket, which is rarely what you want in a bucket that also holds application assets, logs, or deployment artifacts.

The sync schedule uses AWS cron expression syntax with six fields (minutes, hours, day-of-month, month, day-of-week, year). The ON_DEMAND value disables automatic scheduling entirely — use it for data sources where you control sync timing programmatically via StartDataSourceSyncJobCommand. For production MCP servers where document freshness is important, a daily schedule at an off-peak hour (02:00–04:00 UTC) balances freshness against the cost of large sync jobs running during peak Lambda execution hours.

Document metadata JSON format

Kendra's metadata sidecar files extend each indexed document with structured attributes that can be used in AttributeFilter expressions (to scope query results), Facets requests (to build faceted navigation), and DocumentRelevanceOverride (to boost or demote results at query time). The sidecar file for a document at key docs/report-q3.pdf must be placed at docs-metadata/docs/report-q3.pdf.metadata.json — the metadata prefix is prepended to the full document path, and .metadata.json is appended to the document filename.

The metadata JSON has two top-level sections: Attributes (custom attributes matching the DocumentAttributeDefinitions configured on the index) and optional built-in fields prefixed with _ that override Kendra's auto-extracted values. The built-in fields — _source_uri, _title, _last_updated_at, _created_at, _category — control what Kendra displays in search results for the document, independent of its S3 URL or file metadata.

// Example metadata file for: s3://my-bucket/docs/report-q3-2026.pdf
// Metadata file key:         docs-metadata/docs/report-q3-2026.pdf.metadata.json

const metadataFileContent = {
  // Built-in Kendra fields (prefixed with _)
  // These override auto-extracted values from the document itself

  // _source_uri: the URL displayed in search results for this document.
  // Use this to show a human-readable web URL instead of the raw S3 URI.
  "_source_uri": "https://intranet.example.com/docs/finance/report-q3-2026",

  // _title: the document title shown in result items.
  // If omitted, Kendra uses the extracted title from the document header.
  "_title": "Q3 2026 Financial Results — Board Summary",

  // _last_updated_at: ISO 8601 date string; used for recency ranking signals.
  "_last_updated_at": "2026-11-01T10:30:00Z",

  // _created_at: document creation date.
  "_created_at": "2026-11-01T08:00:00Z",

  // _category: a single string category for the document; shown as a facet.
  "_category": "Financial Reports",

  // Custom attributes — must match DocumentAttributeDefinitions on the index.
  // Type values: STRING_VALUE, STRING_LIST_VALUE, LONG_VALUE, DATE_VALUE

  "Attributes": {
    "product_line": {
      "Value": {
        "StringValue": "enterprise"
      }
    },
    "document_version": {
      "Value": {
        "StringValue": "v2.1"
      }
    },
    "published_at": {
      "Value": {
        "DateValue": "2026-11-01T00:00:00Z"
      }
    },
    "view_count": {
      "Value": {
        "LongValue": 0
      }
    },
    "applicable_regions": {
      "Value": {
        "StringListValue": ["us-east-1", "eu-west-1", "ap-southeast-1"]
      }
    }
  }
};

// TypeScript function to generate and upload metadata files alongside documents

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

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

interface DocumentMetadata {
  sourceUri?: string;      // Human-readable URL to show in results
  title?: string;          // Override for document title
  lastUpdatedAt?: Date;    // Document modification date
  createdAt?: Date;        // Document creation date
  category?: string;       // Top-level category for faceting
  customAttributes?: Record;
}

async function uploadDocumentWithMetadata(params: {
  bucket: string;
  documentPrefix: string;    // e.g. 'docs/'
  metadataPrefix: string;    // e.g. 'docs-metadata/docs/'  — must end with /
  documentKey: string;       // relative to documentPrefix, e.g. 'finance/report-q3.pdf'
  documentBody: Buffer;
  contentType: string;
  metadata: DocumentMetadata;
}): Promise {
  const fullDocumentKey = `${params.documentPrefix}${params.documentKey}`;
  const fullMetadataKey = `${params.metadataPrefix}${params.documentKey}.metadata.json`;

  // Build the metadata JSON structure
  const metadataBody: Record = {};

  if (params.metadata.sourceUri) {
    metadataBody['_source_uri'] = params.metadata.sourceUri;
  }
  if (params.metadata.title) {
    metadataBody['_title'] = params.metadata.title;
  }
  if (params.metadata.lastUpdatedAt) {
    metadataBody['_last_updated_at'] = params.metadata.lastUpdatedAt.toISOString();
  }
  if (params.metadata.createdAt) {
    metadataBody['_created_at'] = params.metadata.createdAt.toISOString();
  }
  if (params.metadata.category) {
    metadataBody['_category'] = params.metadata.category;
  }

  if (params.metadata.customAttributes) {
    const attributes: Record = {};
    for (const [key, attr] of Object.entries(params.metadata.customAttributes)) {
      const valueKey = attr.type === 'STRING_VALUE'      ? 'StringValue'
                     : attr.type === 'STRING_LIST_VALUE' ? 'StringListValue'
                     : attr.type === 'LONG_VALUE'        ? 'LongValue'
                     : 'DateValue';

      const rawValue = attr.type === 'DATE_VALUE' && attr.value instanceof Date
        ? attr.value.toISOString()
        : attr.value;

      attributes[key] = { Value: { [valueKey]: rawValue } };
    }
    metadataBody['Attributes'] = attributes;
  }

  // Upload document first
  await s3Client.send(new PutObjectCommand({
    Bucket: params.bucket,
    Key: fullDocumentKey,
    Body: params.documentBody,
    ContentType: params.contentType,
  }));

  // Upload metadata sidecar
  await s3Client.send(new PutObjectCommand({
    Bucket: params.bucket,
    Key: fullMetadataKey,
    Body: JSON.stringify(metadataBody, null, 2),
    ContentType: 'application/json',
  }));

  console.log(`Uploaded document: s3://${params.bucket}/${fullDocumentKey}`);
  console.log(`Uploaded metadata: s3://${params.bucket}/${fullMetadataKey}`);
}

Metadata attributes are indexed during the sync job that follows the sidecar file upload. If you upload a metadata sidecar for a document that was already indexed in a previous sync, Kendra will re-index the document on the next sync run to pick up the new attributes — but only if the document file itself has changed (Kendra uses content hashes for change detection). To force re-indexing of metadata without changing the document body, trigger a full resync (see the incremental sync section below) or use the BatchPutDocument API to push the document and metadata directly without going through the S3 connector sync cycle.

ACL mapping for user-scoped search

Kendra's access control system restricts which documents a given user can see in search results based on principals (users and groups) listed in each document's metadata. When an MCP tool calls QueryCommand with a UserContext object specifying the requesting user's ID and group memberships, Kendra filters the result set to only include documents whose _acl_principals metadata allows access for that user or any of their groups. Documents with no ACL metadata are visible to all users.

ACL filtering is enforced server-side by Kendra — it is not a post-query filter applied in your MCP handler code. This is an important security distinction: even if your MCP server code has access to query Kendra, users will only see documents they are authorized to access, as long as ACL metadata is correctly maintained and UserContext is passed on every query. ACL filtering requires ENTERPRISE_EDITION — on DEVELOPER_EDITION, the UserContext field is accepted by the API but silently ignored and all documents are returned regardless of their ACL metadata.

// ACL metadata format for a document's .metadata.json sidecar
// This restricts the document to members of 'engineering@example.com'
// and the specific user 'alice@example.com'

const aclMetadataExample = {
  "_source_uri": "https://intranet.example.com/docs/engineering/api-spec.html",
  "_title": "Internal API Specification v3.2",
  "_acl_principals": [
    // Allow an entire group
    {
      "Name": "engineering@example.com",
      "Type": "GROUP",
      "Access": "ALLOW"
    },
    // Allow a specific user outside the group
    {
      "Name": "alice@example.com",
      "Type": "USER",
      "Access": "ALLOW"
    },
    // Explicitly deny a user even if they are in an ALLOW group
    // (DENY takes precedence over ALLOW)
    {
      "Name": "contractor-bob@example.com",
      "Type": "USER",
      "Access": "DENY"
    }
  ],
  "Attributes": {
    "product_line": {
      "Value": { "StringValue": "platform" }
    }
  }
};

// QueryCommand with UserContext for ACL-filtered results
import {
  KendraClient,
  QueryCommand,
  type UserContext,
} from '@aws-sdk/client-kendra';

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

const INDEX_ID = process.env.KENDRA_INDEX_ID!;

async function queryWithUserContext(params: {
  queryText: string;
  userId: string;        // e.g. 'alice@example.com' — matches USER principal names
  groups: string[];      // e.g. ['engineering@example.com', 'all-staff@example.com']
  pageSize?: number;
}) {
  const userContext: UserContext = {
    UserId: params.userId,
    Groups: params.groups,
    // DataSourceGroups can be used for data source-specific group mappings
    // when using an identity data source connector (Okta, IAM Identity Center)
    // DataSourceGroups: [{ GroupId: 'eng-team', DataSourceId: 'ds-abc123' }],
  };

  const response = await kendraClient.send(new QueryCommand({
    IndexId: INDEX_ID,
    QueryText: params.queryText,
    PageSize: params.pageSize ?? 10,
    UserContext: userContext,
    // UserContextPolicy must be set to ATTRIBUTE_FILTER on the index
    // (configured via CfnIndex.userContextPolicy = 'ATTRIBUTE_FILTER')
  }));

  return response.ResultItems ?? [];
}

// MCP tool handler integrating user identity from the MCP request context
export async function handleSearchWithContext(args: {
  query: string;
  userId: string;
  groups: string[];
  maxResults?: number;
}) {
  // Validate that required identity fields are present
  if (!args.userId) {
    return {
      content: [{
        type: 'text' as const,
        text: 'Error: userId is required for access-controlled search.',
      }],
      isError: true,
    };
  }

  const results = await queryWithUserContext({
    queryText: args.query,
    userId: args.userId,
    groups: args.groups,
    pageSize: args.maxResults ?? 10,
  });

  if (results.length === 0) {
    return {
      content: [{
        type: 'text' as const,
        text: `No results found for "${args.query}" with your current permissions.`,
      }],
    };
  }

  const formatted = results.map(item => {
    const title = item.DocumentTitle?.Text ?? 'Untitled';
    const excerpt = item.DocumentExcerpt?.Text ?? '';
    const uri = item.DocumentURI ?? '';
    return `### ${title}\n${excerpt}\n${uri ? `_Source: ${uri}_` : ''}`;
  }).join('\n\n');

  return {
    content: [{ type: 'text' as const, text: formatted }],
  };
}

The identity data source integration path — connecting Kendra to Okta or IAM Identity Center to automatically sync group memberships into Kendra — eliminates the need to pass Groups in every QueryCommand call. With an identity data source, Kendra resolves group memberships itself from the identity provider at query time, and your MCP handler only needs to provide UserId. Without an identity data source, your MCP handler is responsible for passing the correct group list; this list must come from a trusted source (the authenticated session, a JWT claim verified by your auth layer) rather than from user-supplied input.

Incremental sync vs full resync

Kendra's S3 connector tracks the content hash of every indexed document. On an incremental sync (the default), documents whose content hash has not changed since the last sync are skipped — Kendra only processes new, modified, and deleted documents. This makes incremental syncs fast and cheap for large corpora where only a small fraction of documents change between sync runs. The tradeoff is that metadata-only changes (sidecar file updated, document body unchanged) are not picked up by incremental syncs, because the document hash is unchanged.

A full resync (called FULL_CRAWL in the API) ignores the hash cache and processes every document in the configured prefix, re-extracting text and re-indexing all metadata. Use a full resync when: you have added new metadata sidecar files to previously indexed documents; you have changed the DocumentAttributeDefinitions on the index (new attributes do not backfill automatically); or you suspect the incremental sync state has drifted from the actual S3 content due to sync job failures.

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

const kendraClient = 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!;

// Trigger an incremental sync (default — only processes changed documents)
async function startIncrementalSync(): Promise {
  const response = await kendraClient.send(new StartDataSourceSyncJobCommand({
    Id: DATA_SOURCE_ID,
    IndexId: INDEX_ID,
  }));
  return response.ExecutionId ?? '';
}

// Trigger a full resync — processes every document in the S3 prefix
// Note: FULL_CRAWL is not a parameter on StartDataSourceSyncJob in the standard API.
// To force a full resync, delete the data source and recreate it, OR use
// the BatchPutDocument API to re-ingest documents directly.
//
// For the S3 connector specifically, the workaround is to update the data source
// configuration (changing any field, e.g. description) which resets the sync state:
async function forceFullResync(newDescription: string): Promise {
  // Updating the data source resets Kendra's hash cache for this connector
  const { UpdateDataSourceCommand } = await import('@aws-sdk/client-kendra');
  await kendraClient.send(new UpdateDataSourceCommand({
    Id: DATA_SOURCE_ID,
    IndexId: INDEX_ID,
    Description: newDescription,
  }));
  // Now trigger a sync — Kendra will treat all documents as new
  await startIncrementalSync();
}

// Poll sync job status — useful in MCP tools that need to verify data freshness
async function waitForSyncCompletion(
  executionId: string,
  timeoutMs = 300_000,
  pollIntervalMs = 15_000,
): Promise {
  const deadline = Date.now() + timeoutMs;

  while (Date.now() < deadline) {
    const response = await kendraClient.send(new ListDataSourceSyncJobsCommand({
      Id: DATA_SOURCE_ID,
      IndexId: INDEX_ID,
      MaxResults: 10,
    }));

    const job = (response.History ?? []).find(
      h => h.ExecutionId === executionId,
    );

    if (!job) {
      await new Promise(resolve => setTimeout(resolve, pollIntervalMs));
      continue;
    }

    const status = job.Status;
    console.log(`Sync job ${executionId}: ${status}`);
    console.log(`  Documents added:   ${job.Metrics?.DocumentsAdded ?? 0}`);
    console.log(`  Documents modified:${job.Metrics?.DocumentsModified ?? 0}`);
    console.log(`  Documents deleted: ${job.Metrics?.DocumentsDeleted ?? 0}`);
    console.log(`  Documents failed:  ${job.Metrics?.DocumentsFailed ?? 0}`);

    if (status === 'SUCCEEDED' || status === 'FAILED' || status === 'INCOMPLETE') {
      return status as DataSourceSyncJobStatus;
    }

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

  throw new Error(`Sync job ${executionId} did not complete within ${timeoutMs}ms`);
}

// MCP tool: check last successful sync and trigger one if stale
export async function ensureDataFreshness(maxAgeMinutes = 60): Promise<{
  fresh: boolean;
  lastSyncTime?: Date;
  status?: string;
}> {
  const response = await kendraClient.send(new ListDataSourceSyncJobsCommand({
    Id: DATA_SOURCE_ID,
    IndexId: INDEX_ID,
    MaxResults: 5,
    StatusFilter: 'SUCCEEDED',
  }));

  const lastSuccess = response.History?.[0];

  if (!lastSuccess?.EndTime) {
    // No successful sync on record — trigger one
    const execId = await startIncrementalSync();
    return { fresh: false, status: `Sync triggered: ${execId}` };
  }

  const ageMs = Date.now() - lastSuccess.EndTime.getTime();
  const ageMins = ageMs / 60_000;

  if (ageMins > maxAgeMinutes) {
    const execId = await startIncrementalSync();
    return {
      fresh: false,
      lastSyncTime: lastSuccess.EndTime,
      status: `Data stale (${Math.round(ageMins)}min old). Sync triggered: ${execId}`,
    };
  }

  return {
    fresh: true,
    lastSyncTime: lastSuccess.EndTime,
    status: `Data fresh (${Math.round(ageMins)}min old)`,
  };
}

The sync job metrics — DocumentsAdded, DocumentsModified, DocumentsDeleted, DocumentsFailed — are the primary signal for sync health monitoring. A sync that completes with Status: INCOMPLETE (rather than SUCCEEDED) means some documents failed to index — typically because they hit a file size limit, contained unsupported content, or had malformed metadata JSON. The detailed error list for each failed document is available in the CloudWatch log group /aws/kendra/INDEX_ID in the stream for the specific sync job execution. For an MCP tool that depends on complete indexing, treat INCOMPLETE syncs as alerts requiring investigation and emit a CloudWatch metric from your handler when the sync status check returns INCOMPLETE.

Supported file types and limits

Kendra's text extraction layer handles the most common enterprise document formats. The file size limits are per-document, not per-corpus — a single PDF over 5 MB is rejected regardless of how many documents are in the index. For documents that exceed these limits, the recommended approach is to pre-process them: split large PDFs with a library like pdf-lib, convert large Word documents to multiple smaller files, or extract text to .txt format before uploading to the S3 source prefix.

Files in unsupported formats (CSV, JSON, XML, ZIP archives, images) are silently skipped during sync — the sync job completes successfully and the skipped files do not appear in the failed document count. The only way to detect format-based exclusions is to compare the number of objects matching the configured prefix in S3 against the DocumentsAdded count in the sync job metrics. Set up a CloudWatch alarm on the difference: if S3ObjectCount - DocumentsAdded - DocumentsFailed > threshold, some documents are being silently skipped due to unsupported formats or exclusion patterns.

Common failures