Amazon Neptune for MCP Servers: Graph Database, Gremlin Traversals, Knowledge Graph Schema, and Bulk Loading

Part 1: Neptune cluster architecture, IAM auth, and gremlin-javascript connection patterns

Neptune runs a fully managed graph cluster with shared cluster storage. High availability is a storage-layer guarantee — data replicates across three Availability Zones regardless of instance count. This is different from DynamoDB (partition-based HA) or RDS Multi-AZ (standby instance HA). The storage layer is always multi-AZ; the question is only how many instance replicas you provision.

Cluster endpoint types

Every Neptune cluster has two endpoint types:

• Writer endpoint: always routes to the current primary instance. Format: your-cluster.cluster-XXXXXX.us-east-1.neptune.amazonaws.com. Use for all write operations. Can serve reads but adds latency to write-heavy workloads — prefer the reader endpoint for read-only MCP tools.
• Reader endpoint: load-balances across all healthy reader instances. Format: your-cluster.cluster-ro-XXXXXX.us-east-1.neptune.amazonaws.com. Lag behind writer is typically under 100 ms. If you have zero reader instances, the reader endpoint fails over to the writer.

Instance sizing: db.r6g.large (4 vCPU, 32 GiB) for dev/test; db.r6g.xlarge (8 vCPU, 64 GiB) for small production; db.r6g.4xlarge (16 vCPU, 128 GiB) for medium production. Neptune uses memory for query caching and buffer pool — size for your working set, not raw data size. Neptune Serverless is an alternative for unpredictable workloads: set minCapacity: 1 and maxCapacity: 32 NCUs; note the ~5–10 s cold start after idle periods, which makes it unsuitable for latency-sensitive real-time MCP tool calls.

CDK cluster setup with IAM authentication

Always enable IAM database authentication on new clusters. It eliminates database passwords and integrates cleanly with Lambda execution roles.

import * as neptune from '@aws-cdk/aws-neptune-alpha';
import * as ec2 from 'aws-cdk-lib/aws-ec2';
import * as iam from 'aws-cdk-lib/aws-iam';
import { Stack, StackProps, RemovalPolicy } from 'aws-cdk-lib';

export class McpNeptuneStack extends Stack {
  constructor(scope: Construct, id: string, props: StackProps) {
    super(scope, id, props);

    const vpc = ec2.Vpc.fromLookup(this, 'Vpc', { isDefault: false });

    const cluster = new neptune.DatabaseCluster(this, 'McpGraph', {
      vpc,
      instanceType: neptune.InstanceType.R6G_LARGE,
      iamAuthentication: true,  // required — no passwords
      removalPolicy: RemovalPolicy.SNAPSHOT,
      cloudwatchLogsExports: [
        neptune.LogType.AUDIT,
        neptune.LogType.SLOWQUERY,
      ],
      parameterGroup: new neptune.ClusterParameterGroup(this, 'ClusterParams', {
        parameters: {
          neptune_enable_audit_log: '1',
          neptune_query_timeout: '30000',  // 30 s; default 120 s
          neptune_result_cache: '1',       // in-memory query caching
          neptune_streams: '1',            // enable change data capture
        },
      }),
    });

    // Allow Lambda execution role to connect
    const mcpLambdaRole = iam.Role.fromRoleArn(this, 'McpRole', 'arn:aws:iam::ACCOUNT:role/McpLambdaRole');
    cluster.grantConnect(mcpLambdaRole);
    // grantConnect adds neptune-db:connect on the cluster ARN
  }
}

cluster.grantConnect() attaches neptune-db:connect on the ARN arn:aws:neptune-db:REGION:ACCOUNT:cluster:CLUSTER_ID/*. Without it, Neptune rejects the SigV4-signed connection with an HTTP 403 — a confusing error because the WebSocket connection itself may appear to succeed before the query fails.

Connecting from a Lambda MCP server

Neptune's Gremlin endpoint uses WebSockets over TLS (port 8182, path /gremlin). The neptune-sigv4-signer package signs requests with SigV4 credentials from the Lambda execution role.

import gremlin from 'gremlin';
import { createSigner } from 'neptune-sigv4-signer';

const { driver: { DriverRemoteConnection } } = gremlin;
const { AnonymousTraversalSource } = gremlin.structure.Graph.prototype;
const __ = gremlin.process.statics;
const { T, cardinality, P } = gremlin.process;

const NEPTUNE_ENDPOINT = process.env.NEPTUNE_ENDPOINT!;
const NEPTUNE_PORT = 8182;
const REGION = process.env.AWS_REGION || 'us-east-1';

let g: gremlin.process.GraphTraversalSource | null = null;
let dc: DriverRemoteConnection | null = null;

async function getTraversalSource() {
  // Reconnect if connection is closed or was never opened
  if (g && dc) return g;

  const url = `wss://${NEPTUNE_ENDPOINT}:${NEPTUNE_PORT}/gremlin`;
  const signer = createSigner({ region: REGION });

  dc = new DriverRemoteConnection(url, {
    mimeType: 'application/vnd.gremlin-v3.0+json',
    headers: await signer.getAuthHeaders(url, 'GET'),
    rejectUnauthorized: true,
  });

  g = gremlin.structure.Graph.traversal().withRemote(dc);
  return g;
}

export async function handler(event: McpToolEvent) {
  const g = await getTraversalSource();

  try {
    const result = await g.V().hasLabel('Entity')
      .has('name', event.entityName)
      .valueMap(true)
      .next();
    return result.value;
  } catch (err) {
    // Neptune drops idle connections after ~20 min.
    // Reset and let next invocation reconnect.
    if (dc) { try { await dc.close(); } catch {} }
    dc = null; g = null;
    throw err;
  }
}

Lambda reconnect strategy: Neptune drops idle WebSocket connections after approximately 20 minutes. For Lambda-backed MCP servers, it's safest to create a fresh connection per invocation rather than reusing a module-level connection — the extra ~100 ms connection overhead is preferable to silent failures on stale connections. For long-lived ECS or EC2 deployments, maintain a connection pool with reconnect-on-error logic.

Security group and VPC configuration

// CDK: allow Lambda to reach Neptune on port 8182
const neptuneSecurityGroup = new ec2.SecurityGroup(this, 'NeptuneSg', { vpc });
const lambdaSecurityGroup = new ec2.SecurityGroup(this, 'LambdaSg', { vpc });

neptuneSecurityGroup.addIngressRule(
  ec2.Peer.securityGroupId(lambdaSecurityGroup.securityGroupId),
  ec2.Port.tcp(8182),
  'Allow Gremlin connections from MCP Lambda',
);

const mcpLambda = new lambda.Function(this, 'McpHandler', {
  vpc,
  securityGroups: [lambdaSecurityGroup],
  vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS },
  // ...
});

VPC-attached Lambda adds ~300–500 ms ENI attachment on cold starts. Neptune IAM auth adds another ~100–150 ms per new connection. Plan for 500–700 ms total overhead for cold MCP tool calls that require a fresh Neptune connection. Warm invocations that reuse the module-level connection execute in <10 ms graph round-trips for simple traversals.

Query language selection

For MCP server knowledge graphs, Gremlin is the right default. You can mix Gremlin and openCypher on the same cluster if needed — Neptune runs both on the same property graph engine. SPARQL is only relevant if you are building on a formal ontology or integrating with an existing RDF triple store.

Common Neptune connection failures

Part 2: Gremlin traversal patterns for MCP tool calls

Gremlin is the query language of Apache TinkerPop. It's an imperative graph traversal language — you describe the steps a traversal takes through the graph rather than declaring the result pattern you want. For MCP server developers, the key Gremlin patterns are vertex/edge CRUD, multi-hop traversals, the upsert pattern with coalesce(), paginated queries for LLM tool responses, and batched bulk insert.

Property graph model

Neptune's property graph has four first-class concepts: vertices (nodes), edges (directed relationships), properties (key-value pairs on vertices or edges), and IDs (every vertex and edge has a string ID). Important: Neptune does NOT support multi-label vertices — unlike Neo4j, you cannot do :Entity:Person. Model multiple types with a hierarchy: use a base label (Entity) with a subtype property ('Person').

Basic vertex and edge CRUD

import gremlin from 'gremlin';

const { process: { P, T, cardinality } } = gremlin;
const __ = gremlin.process.statics;

// Add a vertex with a stable string ID
await g.addV('Entity')
  .property(T.id, 'entity::aws-lambda')   // T.id is TinkerPop token for graph ID
  .property('name', 'AWS Lambda')          // user properties
  .property('category', 'compute')
  .property('createdAt', Date.now())
  .next();

// CRITICAL: T.id vs 'id'
// .property(T.id, value)  — sets the GRAPH ID (the vertex's identity in Neptune)
// .property('id', value)  — sets a USER PROPERTY named "id" (a regular property)
// Never confuse them. Using .property('id', ...) does NOT set the vertex ID.

// Add another vertex and an edge
await g.addV('Concept')
  .property(T.id, 'concept::serverless')
  .property('name', 'serverless computing')
  .next();

await g.addE('RELATES_TO')
  .from_(g.V('entity::aws-lambda'))
  .to_(g.V('concept::serverless'))
  .property('weight', 0.9)
  .next();

// Update a property — use cardinality.single to overwrite
await g.V('entity::aws-lambda')
  .property(cardinality.single, 'category', 'compute-serverless')
  .next();
// Without cardinality.single, Neptune uses cardinality.list (default)
// and APPENDS a second value instead of overwriting.
// This produces SchemaViolationException or silent duplicates.

// Read a vertex with all properties and graph ID/label
const vertex = await g.V('entity::aws-lambda').valueMap(true).next();
// valueMap(true) includes T.id and T.label; valueMap() without true omits them.

// Delete a vertex (also removes all incident edges)
await g.V('entity::aws-lambda').drop().next();

Multi-hop traversal patterns

// Find outgoing neighbors
const neighbors = await g.V('entity::aws-lambda')
  .out('RELATES_TO')
  .valueMap(true)
  .toList();

// Find incoming edges — what other entities mention this concept?
const mentioners = await g.V('concept::serverless')
  .in_('RELATES_TO')   // underscore avoids conflict with JS 'in' keyword
  .hasLabel('Entity')
  .values('name')
  .toList();

// Two-hop traversal with dedup
const twoHops = await g.V('entity::aws-lambda')
  .out('RELATES_TO')
  .out('RELATES_TO')
  .dedup()
  .limit(50)          // ALWAYS add limit before toList on multi-hop traversals
  .valueMap(true)
  .toList();

// Depth-limited traversal with cycle prevention
const paths = await g.V('entity::aws-lambda')
  .repeat(__.out('RELATES_TO').simplePath())  // simplePath() prevents revisiting vertices
  .times(3)                                   // maximum 3 hops
  .path()
  .by(__.values('name'))                      // map each step to 'name' property
  .limit(10)
  .toList();

// Shortest path between two vertices
const shortestPath = await g.V('entity::aws-lambda')
  .repeat(__.both().simplePath())
  .until(__.hasId('concept::kubernetes'))
  .path()
  .by(__.values('name'))
  .limit(1)
  .next();

// Project results to structured objects for MCP tool responses
const structured = await g.V()
  .hasLabel('Entity')
  .has('category', 'compute')
  .limit(20)
  .project('id', 'name', 'neighborCount')
  .by(T.id)
  .by(__.values('name'))
  .by(__.both('RELATES_TO').count())
  .toList();

The critical rule: never call .toList() on an unbounded traversal. Neptune executes traversals eagerly — without a .limit(N) before .toList(), a traversal over a large graph returns every matching record before your code sees any results. For traversals that might match millions of vertices, use .limit(), .range(), or .count() first to understand scale.

Upsert pattern with coalesce()

// The idiomatic Neptune upsert: create if not exists, return existing if found
async function upsertEntity(id: string, name: string, category: string) {
  return g.V(id)
    .fold()           // fold()/collect into a list — empty list if vertex not found
    .coalesce(
      __.unfold(),    // if list non-empty: unfold it (return existing vertex)
      __.addV('Entity')                    // if list empty: create new vertex
        .property(T.id, id)
        .property('name', name)
        .property('category', category)
        .property('createdAt', Date.now()),
    )
    .next();
}

// CRITICAL: fold() before coalesce() is MANDATORY.
// Without fold(), coalesce() receives an empty traverser (not an empty list)
// when the vertex doesn't exist, and the second argument (addV) is never evaluated.
// The coalesce() with fold() pattern is: "turn zero-or-one vertices into a
// definitely-one-item list (either the vertex or an empty list), then coalesce
// on the list contents."

// Upsert an edge between two existing vertices
async function upsertEdge(fromId: string, toId: string, label: string, weight: number) {
  return g.V(fromId).as('from')
    .V(toId).as('to')
    .select('from')
    .outE(label).where(__.inV().as('to'))  // find edge from 'from' to 'to'
    .fold()
    .coalesce(
      __.unfold(),
      __.addE(label)
        .from_(__.V(fromId))
        .to_(__.V(toId))
        .property('weight', weight),
    )
    .next();
}

Pagination for MCP tool responses

interface PaginatedResult<T> {
  items: T[];
  nextOffset: number | null;
}

async function listEntities(
  category: string,
  offset: number = 0,
  pageSize: number = 20,
): Promise<PaginatedResult<{ id: string; name: string }>> {
  // Fetch one extra to detect whether there's a next page
  const items = await g.V()
    .hasLabel('Entity')
    .has('category', category)
    .order().by('name', gremlin.process.order.asc)
    .range(offset, offset + pageSize + 1)
    .valueMap(true)
    .toList();

  const hasMore = items.length > pageSize;
  const page = hasMore ? items.slice(0, pageSize) : items;

  return {
    items: page.map(v => ({
      id: v[T.id] as string,
      name: (v['name'] as string[])[0],  // properties come back as arrays in valueMap
    })),
    nextOffset: hasMore ? offset + pageSize : null,
  };
}

// Keyset pagination by ID — more stable than offset for concurrent writes
async function listEntitiesAfterCursor(afterId: string | null, pageSize = 20) {
  let traversal = g.V().hasLabel('Entity').order().by(T.id);
  if (afterId) {
    traversal = traversal.has(T.id, P.gt(afterId));
  }
  return traversal.limit(pageSize).valueMap(true).toList();
}

Bulk insert for knowledge graph ingestion

async function bulkUpsertGraph(entities: EntityRecord[], edges: EdgeRecord[]) {
  // Neptune has a ~65 KB Gremlin script size limit.
  // Chunk to <100 entities per traversal to stay comfortably under it.
  const CHUNK_SIZE = 100;

  // Insert vertices first
  for (let i = 0; i < entities.length; i += CHUNK_SIZE) {
    const chunk = entities.slice(i, i + CHUNK_SIZE);

    // Chain all upserts in one traversal for a single round-trip
    let t: any = g;
    for (const entity of chunk) {
      t = t.V(entity.id)
        .fold()
        .coalesce(
          __.unfold(),
          __.addV(entity.label)
            .property(T.id, entity.id)
            .property('name', entity.name),
        )
        .property(cardinality.single, 'updatedAt', Date.now());

      for (const [k, v] of Object.entries(entity.properties)) {
        t = t.property(cardinality.single, k, v);
      }
    }
    await t.iterate();
  }

  // Insert edges after vertices (avoid foreign-key-equivalent errors)
  for (let i = 0; i < edges.length; i += CHUNK_SIZE) {
    const chunk = edges.slice(i, i + CHUNK_SIZE);
    for (const edge of chunk) {
      await upsertEdge(edge.fromId, edge.toId, edge.label, edge.weight);
    }
  }
}
// For >100K entities, use Neptune bulk load from S3 instead of Gremlin (see Part 4)

Gremlin common failures

Part 3: Knowledge graph schema design for MCP agent memory

The most common Neptune mistake in MCP server deployments is not a code error — it's a schema error. Teams model entities as flat vertices and relationships as generic edges without thinking through what queries the LLM will actually need to run, how facts change over time, or how to isolate data for multiple tenants. The schema decisions made at the start of a project are expensive to change later because they require rewriting all ingestion pipelines and re-loading the entire graph.

When to use graph memory over vector memory

Vector stores (OpenSearch k-NN, Bedrock Knowledge Bases) return semantically similar passages. Property graphs return relationship chains. Use both together:

• Vector search for initial retrieval: "find chunks semantically similar to the user's query."
• Graph traversal for enrichment: "given the entities mentioned in those chunks, what other entities and facts connect to them in the knowledge base?"

Graph enrichment adds relational context that vector search misses. If the user asks "what are the dependencies of service X?", a vector search returns service documentation — but a graph traversal returns the exact dependency chain: X → depends_on → Y → depends_on → Z → hosted_in → us-east-1.

Core schema: vertex labels

Core schema: edge labels

MCP tool definitions for the knowledge graph

// Tool: store_entity — upsert an entity after extraction
{
  name: 'store_entity',
  inputSchema: {
    type: 'object',
    properties: {
      id: { type: 'string', description: 'Stable entity ID, e.g. entity::Service::aws-lambda' },
      name: { type: 'string' },
      entityType: { type: 'string', enum: ['Person', 'Organization', 'Service', 'Concept', 'Location'] },
      description: { type: 'string' },
      sourceDocumentId: { type: 'string' },
    },
    required: ['id', 'name', 'entityType'],
  },
}

// Tool: add_relationship — connect two entities
{
  name: 'add_relationship',
  inputSchema: {
    type: 'object',
    properties: {
      fromId: { type: 'string' },
      toId: { type: 'string' },
      relationshipType: { type: 'string', description: 'Semantic label: depends_on, is_part_of, authored_by, etc.' },
      weight: { type: 'number', minimum: 0, maximum: 1 },
    },
    required: ['fromId', 'toId', 'relationshipType'],
  },
}

// Tool: get_entity_context — retrieve graph neighborhood for RAG enrichment
{
  name: 'get_entity_context',
  inputSchema: {
    type: 'object',
    properties: {
      entityId: { type: 'string' },
      hops: { type: 'integer', minimum: 1, maximum: 3, default: 2 },
      includeSourceChunks: { type: 'boolean', default: true },
    },
    required: ['entityId'],
  },
}

// Implementation of get_entity_context
async function getEntityContext(entityId: string, hops: number, includeSourceChunks: boolean) {
  const g = await getTraversalSource();

  const entity = await g.V(entityId).valueMap(true).next();
  if (!entity.value) throw new Error(`Entity not found: ${entityId}`);

  // N-hop neighborhood with distance (loop count)
  const neighborhood = await g.V(entityId)
    .repeat(__.both('RELATES_TO').simplePath())
    .times(hops)
    .dedup()
    .limit(30)
    .project('entity', 'distance')
    .by(__.valueMap(true))
    .by(__.loops())   // loops() = current traversal depth
    .toList();

  let sourceChunks: string[] = [];
  if (includeSourceChunks) {
    sourceChunks = await g.V(entityId)
      .in_('MENTIONS')               // Chunk that mentions this entity
      .has('tokenCount', P.lt(512))  // keep context short
      .limit(5)
      .values('text')
      .toList() as string[];
  }

  return {
    entity: entity.value,
    neighbors: neighborhood.map(n => ({ entity: n.entity, hops: n.distance })),
    sourceChunks,
  };
}

Temporal fact versioning

Facts in a knowledge graph change over time. Neptune has no native time-travel feature — model temporal validity with validFrom and validTo epoch-ms properties.

// Add a versioned fact
await g.addV('Fact')
  .property(T.id, `fact::${hash(claim)}`)
  .property('claim', claim)
  .property('confidence', 0.95)
  .property('validFrom', Date.now())
  // No validTo = currently valid. Neptune represents null properties by ABSENCE.
  // has('validTo', null) matches vertices where the property is absent.
  .next();

// Supersede an old fact with a new one
async function supersedeFact(oldFactId: string, newClaim: string) {
  const now = Date.now();

  // Mark old fact expired
  await g.V(oldFactId)
    .property(cardinality.single, 'validTo', now)
    .next();

  // Create new fact
  const newFactId = `fact::${hash(newClaim)}`;
  await g.addV('Fact')
    .property(T.id, newFactId)
    .property('claim', newClaim)
    .property('validFrom', now)
    .next();

  // Link them for provenance
  await g.addE('SUPERSEDES')
    .from_(g.V(newFactId))
    .to_(g.V(oldFactId))
    .property('supersededAt', now)
    .next();
}

// Query only currently valid facts
// has('validTo', null) matches vertices WHERE 'validTo' PROPERTY IS ABSENT
// — this works correctly in Neptune; a vertex without a validTo property
//   matches the has('validTo', null) predicate.
const currentFacts = await g.V()
  .hasLabel('Fact')
  .has('validTo', null)
  .limit(100)
  .values('claim')
  .toList();

Multi-tenant namespace isolation

const TENANT_ID = process.env.TENANT_ID!;  // UUID, injected at deploy time

function tenantVertexId(type: string, name: string): string {
  return `${TENANT_ID}::${type}::${createHash('sha256').update(name).digest('hex').slice(0, 16)}`;
}

// All graph operations prefix IDs with tenant
async function storeEntityForTenant(name: string, entityType: string) {
  const id = tenantVertexId('entity', `${entityType}::${name.toLowerCase()}`);
  return upsertEntity(id, name, entityType);
}

// Traversals scope to tenant's vertex ID prefix using P.between()
async function getEntitiesForTenant(entityType: string, limit = 20) {
  return g.V()
    .hasLabel('Entity')
    .has(T.id, P.between(
      `${TENANT_ID}::entity::`,    // inclusive lower bound
      `${TENANT_ID}::entity::z`,   // exclusive upper bound (z sorts after all hex chars)
    ))
    .has('entityType', entityType)
    .limit(limit)
    .valueMap(true)
    .toList();
  // Neptune indexes vertex IDs — P.between() on T.id uses the index efficiently.
}

// SECURITY: validate the tenant prefix on every read
function assertTenantOwns(vertexId: string) {
  if (!vertexId.startsWith(`${TENANT_ID}::`)) {
    throw new Error(`Access denied: vertex ${vertexId} does not belong to tenant ${TENANT_ID}`);
  }
}

Neptune has no row-level security equivalent. Isolation is entirely by ID prefix convention enforced at the application layer. Use UUID tenant IDs (not organization names, which could be guessed) as prefixes. The assertTenantOwns() check is not just defensive programming — it's the only isolation boundary between tenants.

Schema common failures

Part 4: Neptune bulk load from S3 and Streams CDC

Gremlin-based vertex insertion (from Part 2) is best for incremental updates of hundreds to thousands of entities per MCP tool call. For initial population of a knowledge graph with millions of vertices and edges — loading a documentation corpus, a dependency graph, or a historical entity dataset — the Neptune Loader API is orders of magnitude faster. It reads directly from S3 in parallel, bypassing the Gremlin endpoint entirely, and can load a billion edges in under an hour on appropriately sized instances. Neptune Streams then provides a change log for all subsequent mutations.

When to use bulk load vs Gremlin insert

Gremlin CSV file formats

The bulk loader expects separate CSV files for vertices and edges. The header row controls how Neptune parses each column.

# vertices.csv — vertex file header format
# ~id,~label,propertyName:datatype,...
# datatype options: String, Long, Double, Bool, Date
# Cardinality suffix: propertyName:String(set) for multi-valued, :String(single) for single

~id,~label,name:String,entityType:String,confidence:Double,createdAt:Long
"entity::aws::lambda","Entity","AWS Lambda","Service",0.99,1699430400000
"entity::aws::dynamodb","Entity","Amazon DynamoDB","Service",0.99,1699430400000
"concept::serverless","Concept","serverless computing","Concept",1.0,1699430400000

# edges.csv — edge file header format
# ~id,~from,~to,~label,propertyName:datatype,...
# ~from and ~to MUST reference vertex IDs present in vertex files
# If a referenced vertex ID doesn't exist: edge silently skipped (unless failOnError:TRUE)

~id,~from,~to,~label,relationshipType:String,weight:Double,validFrom:Long
"edge::001","entity::aws::lambda","concept::serverless","RELATES_TO","uses",0.95,1699430400000
"edge::002","entity::aws::dynamodb","concept::nosql","RELATES_TO","implements",0.98,1699430400000
"edge::003","entity::aws::lambda","entity::aws::dynamodb","RELATES_TO","integrates_with",0.90,1699430400000

Important column format details: Date columns use ISO 8601 (2026-11-08T00:00:00Z), not epoch-ms. For epoch timestamps, use Long type with integer milliseconds. Set cardinality columns use semicolons to separate multiple values: "tag1;tag2;tag3".

IAM prerequisites for the bulk loader

// CDK: IAM role for Neptune bulk load — TRUST rds.amazonaws.com, NOT neptune.amazonaws.com
const neptuneS3Role = new iam.Role(this, 'NeptuneS3LoadRole', {
  assumedBy: new iam.ServicePrincipal('rds.amazonaws.com'),  // rds, NOT neptune!
  description: 'Neptune bulk loader S3 read access',
});

// S3 bucket must be in same region as Neptune cluster
neptuneS3Role.addToPolicy(new iam.PolicyStatement({
  actions: ['s3:GetObject', 's3:ListBucket', 's3:GetBucketLocation'],
  resources: [graphDataBucket.bucketArn, graphDataBucket.arnForObjects('*')],
}));

// Associate the role with the Neptune cluster
const cluster = new neptune.DatabaseCluster(this, 'McpGraph', {
  // ...
  associatedRoles: [neptuneS3Role],
});

The rds.amazonaws.com trust principal (not neptune.amazonaws.com) is the most common IAM misconfiguration for Neptune bulk loads. The load job fails immediately with an AccessDenied error when this is wrong. Verify with: aws neptune describe-db-clusters --query 'DBClusters[*].AssociatedRoles'.

Submitting and monitoring a load job

async function submitBulkLoad(s3Prefix: string, roleArn: string): Promise<string> {
  const body = JSON.stringify({
    source: s3Prefix,            // e.g. s3://your-bucket/graph-data/
    format: 'csv',
    iamRoleArn: roleArn,
    region: REGION,
    failOnError: 'FALSE',        // skip bad records; use TRUE in test loads
    parallelism: 'MEDIUM',       // LOW | MEDIUM | HIGH | OVERSUBSCRIBE
    updateSingleCardinalityProperties: 'TRUE',  // overwrite existing properties
    queueRequest: 'TRUE',        // queue if another load is running
  });

  const response = await signedRequest('POST', `https://${NEPTUNE_ENDPOINT}:8182/loader`, body);
  return response.payload.loadId;
}

async function pollLoadStatus(loadId: string): Promise<void> {
  for (let attempt = 0; attempt < 120; attempt++) {
    await new Promise(r => setTimeout(r, 30_000));  // poll every 30 s

    const status = await signedRequest(
      'GET',
      `https://${NEPTUNE_ENDPOINT}:8182/loader/${loadId}?details=TRUE`,
    );
    const { overallStatus, errors } = status.payload;

    if (overallStatus.status === 'LOAD_COMPLETED') {
      console.log(`Loaded ${overallStatus.totalRecords} records in ${overallStatus.elapsedSeconds}s`);
      return;
    }
    if (overallStatus.status === 'LOAD_COMPLETED_WITH_ERRORS') {
      // Some records skipped — fetch error details
      const details = await signedRequest(
        'GET',
        `https://${NEPTUNE_ENDPOINT}:8182/loader/${loadId}?details=TRUE&errors=TRUE`,
      );
      console.warn('Load completed with errors:', details.payload.errors);
      return;
    }
    if (overallStatus.status.startsWith('LOAD_FAILED') ||
        overallStatus.status === 'LOAD_CANCELLED_BY_USER') {
      throw new Error(`Neptune load failed: ${overallStatus.status} — ${JSON.stringify(errors)}`);
    }
    console.log(`Loading: ${overallStatus.fullLoadProgressPercent}% — ${overallStatus.status}`);
  }
  throw new Error('Load timed out after 60 minutes');
}

Load status values: LOAD_NOT_STARTED, LOAD_IN_QUEUE, LOAD_IN_PROGRESS, LOAD_COMPLETED, LOAD_COMPLETED_WITH_ERRORS (some records skipped — fetch error details with errors=TRUE), LOAD_FAILED, LOAD_CANCELLED_BY_USER, LOAD_UNEXPECTED_ERROR. For large initial loads, use parallelism: 'HIGH' and split CSV files to 1–10 GB each for maximum throughput. The S3 prefix must have a trailing slash (s3://bucket/prefix/) to load all files in the prefix in parallel.

Neptune Streams for change data capture

Neptune Streams is a change log that records every graph mutation in order. Enable it with the cluster parameter neptune_streams=1 (requires a cluster reboot). Streams are available via HTTP at GET /gremlin/stream. Each record contains a commit number, operation type (ADD/REMOVE), entity type (vl=vertex label, vp=vertex property, e=edge, ep=edge property), and the data.

// Lambda poller: reads Neptune Streams and publishes to EventBridge
import { DynamoDBClient, GetItemCommand, PutItemCommand } from '@aws-sdk/client-dynamodb';
import { EventBridgeClient, PutEventsCommand } from '@aws-sdk/client-eventbridge';

const ddb = new DynamoDBClient({});
const eb = new EventBridgeClient({});
const NEPTUNE_STREAM_URL = `https://${process.env.NEPTUNE_ENDPOINT}:8182/gremlin/stream`;

async function getLastCommitNum(): Promise<number> {
  const result = await ddb.send(new GetItemCommand({
    TableName: process.env.CURSOR_TABLE!,
    Key: { pk: { S: 'neptune-streams-cursor' } },
  }));
  return result.Item?.commitNum ? parseInt(result.Item.commitNum.N!) : 1;
}

export async function handler() {
  const lastCommit = await getLastCommitNum();

  const response = await signedRequest(
    'GET',
    `${NEPTUNE_STREAM_URL}?commitNum=${lastCommit}&limit=100&iteratorType=AFTER_SEQUENCE_NUMBER`,
  );

  const { records, lastEventId } = response;
  if (!records || records.length === 0) return;

  // Emit EventBridge events for new vertices
  const events = records
    .filter((r: any) => r.type === 'vl' && r.op === 'ADD')
    .map((r: any) => ({
      EventBusName: 'default',
      Source: 'alivemcp.neptune',
      DetailType: 'EntityCreated',
      Detail: JSON.stringify({
        entityId: r.data.id,
        label: r.data.value,
        commitNum: r.commitTimestamp,
      }),
    }));

  if (events.length > 0) {
    await eb.send(new PutEventsCommand({ Entries: events }));
  }

  // Advance cursor to last processed commit
  await ddb.send(new PutItemCommand({
    TableName: process.env.CURSOR_TABLE!,
    Item: { pk: { S: 'neptune-streams-cursor' }, commitNum: { N: lastEventId.commitNum.toString() } },
  }));
}

Streams limitations to be aware of: only available on Neptune DB (not Neptune Serverless), retention is 7 days by default, records are not ordered within a millisecond window (use commitNum + opNum for total ordering), and Streams only captures mutations that occur after enabling — historical data is not in the stream. For multi-region replication, combine Streams with S3 export and cross-region copy.

Bulk load common failures

Production checklist for Neptune-backed MCP servers

1. IAM auth enabled on cluster — iamAuthentication: true in CDK; neptune-db:connect on Lambda execution role via cluster.grantConnect()
2. VPC and security group — Lambda in same VPC as Neptune; Neptune SG has TCP 8182 inbound from Lambda SG; Lambda SG has TCP 443 outbound for STS (IAM auth credential refresh)
3. Reconnect-on-error handling — try/catch in every traversal that resets the connection on error; never assume a module-level connection is still open after 20+ minutes idle
4. Cardinality discipline — all property updates use cardinality.single for single-valued properties; never use default cardinality.list for properties that should be unique per vertex
5. Traversal limits — every traversal has a .limit(N) before .toList(); every recursive repeat() has both .until() and .times(N)
6. Schema designed upfront — vertex labels, edge labels, ID format, tenant isolation strategy, and temporal validity model decided before first write; schema changes require re-loading the entire graph
7. Temporal validity model — validFrom/validTo epoch-ms on edges and Fact vertices; SUPERSEDES edges link new and old facts; queries filter with has('validTo', null) for currently valid facts
8. Multi-tenant isolation — UUID tenant ID prefix on all vertex IDs; assertTenantOwns() validation on all reads; P.between() range scan on T.id for efficient per-tenant queries
9. Bulk load IAM — loader role trusts rds.amazonaws.com (not neptune); role associated with cluster; S3 bucket in same region as Neptune
10. Neptune Streams enabled — neptune_streams=1 in cluster parameter group; Lambda poller with DynamoDB cursor for at-least-once processing; EventBridge for downstream consumers
11. Query timeout tuned — neptune_query_timeout: 30000 ms in production to surface slow queries early; default 120,000 ms allows runaway traversals to hold connections for two minutes
12. Monitoring — Neptune CloudWatch metrics: GremlinRequestsPerSec, GremlinErrors, BufferCacheHitRatio (target >90%), CPUUtilization; alarm on GremlinErrors spike and BufferCacheHitRatio drop below 80%
13. AliveMCP endpoint monitoring — if your Neptune-backed MCP server is public, register its endpoint at AliveMCP so you get alerted the moment a Gremlin connection failure causes a tool call to fail before your users do