Mongodb Atlas Stream Processing

MongoDB Atlas Streams

Build, operate, and debug Atlas Stream Processing (ASP) pipelines using four MCP tools from the MongoDB MCP Server.

Prerequisites

This skill requires the MongoDB MCP Server connected with:

• Atlas API credentials (apiClientId and apiClientSecret)

The 4 tools: atlas-streams-discover, atlas-streams-build, atlas-streams-manage, atlas-streams-teardown.

All operations require an Atlas project ID. If unknown, call atlas-list-projects first to find your project ID.

If MCP tools are unavailable

If the MongoDB MCP Server is not connected or the streams tools are missing, see references/mcp-troubleshooting.md for diagnostic steps and fallback options.

Tool Selection Matrix

atlas-streams-discover — ALL read operations

Pagination (all list actions): limit (1-100, default 20), pageNum (default 1). Response format: responseFormat — "concise" (default for list actions) or "detailed" (default for inspect/diagnose).

atlas-streams-build — ALL create operations

Field mapping — only fill fields for the selected resource type:

• resource = "workspace": Fill: projectId, workspaceName, cloudProvider, region, tier, includeSampleData. Leave empty: all connection and processor fields.
• resource = "connection": Fill: projectId, workspaceName, connectionName, connectionType, connectionConfig. Leave empty: all workspace and processor fields. (See references/connection-configs.md for type-specific schemas.)
• resource = "processor": Fill: projectId, workspaceName, processorName, pipeline, dlq (recommended), autoStart (optional). Leave empty: all workspace and connection fields. (See references/pipeline-patterns.md for pipeline examples.)
• resource = "privatelink": Fill: projectId, privateLinkConfig. Note: PrivateLink is project-level, not workspace-level. workspaceName is not required — omit it. Leave empty: all connection and processor fields.

atlas-streams-manage — ALL update/state operations

Field mapping — always fill projectId, workspaceName, then by action:

• "start-processor" → resourceName. Optional: tier, resumeFromCheckpoint, startAtOperationTime (ISO 8601 timestamp to resume from a specific point)
• "stop-processor" → resourceName
• "modify-processor" → resourceName. At least one of: pipeline, dlq, newName
• "update-workspace" → newRegion or newTier
• "update-connection" → resourceName, connectionConfig. Exception: networking config (e.g., PrivateLink) cannot be modified after creation — delete and recreate.
• "accept-peering" → peeringId, requesterAccountId, requesterVpcId
• "reject-peering" → peeringId

State pre-checks:

• start-processor → errors if processor is already STARTED
• stop-processor → no-ops if already STOPPED or CREATED (not an error)
• modify-processor → errors if processor is STARTED (must stop first)

Processor states: CREATED → STARTED (via start) → STOPPED (via stop). Can also enter FAILED on runtime errors. Modify requires STOPPED or CREATED state.

Teardown safety checks:

• Processor deletion → auto-stops before deleting (no need to stop manually first)
• Connection deletion → blocks if any running processor references it. Stop/delete referencing processors first.
• Workspace deletion → See detailed workflow below (lines 108-111).

atlas-streams-teardown — ALL delete operations

Field mapping — always fill projectId, resource, then:

• resource: "workspace" → workspaceName
• resource: "connection" or "processor" → workspaceName, resourceName
• resource: "privatelink" or "peering" → resourceName (the ID). These are project-level resources, not tied to a specific workspace.

Before deleting a workspace, inspect it first: 1. atlas-streams-discover → inspect-workspace — get connection/processor counts 2. Present to user: "Workspace X contains N connections and M processors. Deleting permanently removes all. Proceed?" 3. Wait for confirmation before calling atlas-streams-teardown

CRITICAL: Validate Before Creating Processors

You MUST call `search-knowledge` before composing any processor pipeline. This is not optional.

• Field validation: Query with the sink/source type, e.g. "Atlas Stream Processing $emit S3 fields" or "Atlas Stream Processing Kafka $source configuration". This catches errors like prefix vs path for S3 $emit.
• Pattern examples: Query with dataSources: [{"name": "devcenter"}] for working pipelines, e.g. "Atlas Stream Processing tumbling window example".

Also fetch examples from the official ASP examples repo when building non-trivial processors: https://github.com/mongodb/ASP_example (quickstarts, example processors, Terraform examples). Start with example_processors/README.md for the full pattern catalog.

Key quickstarts:

Pipeline Rules & Warnings

Invalid constructs — these are NOT valid in streaming pipelines:

• `$$NOW`, `$$ROOT`, `$$CURRENT` — NOT available in stream processing. NEVER use these. Use the document's own timestamp field or _stream_meta metadata for event time instead of $$NOW.
• HTTPS connections as `$source` — HTTPS is for $https enrichment or sink only, NOT as a data source
• Kafka `$source` without `topic` — topic field is required
• Pipelines without a sink — terminal stage ($merge, $emit, $https, or $externalFunction async) required for deployed processors (sinkless only works via sp.process())
• Lambda as `$emit` target — Lambda uses $externalFunction (mid-pipeline enrichment), not $emit
• `$validate` with `validationAction: "error"` — crashes processor; use "dlq" instead

Required fields by stage:

• `$source` (change stream): include fullDocument: "updateLookup" to get the full document content
• `$source` (Kinesis): use stream (NOT streamName or topic)
• `$emit` (Kinesis): MUST include partitionKey
• `$emit` (S3): use path (NOT prefix)
• `$https`: must include connectionName, path, method, as, onError: "dlq"
• `$externalFunction`: must include connectionName, functionName, execution, as, onError: "dlq"
• `$validate`: must include validator with $jsonSchema and validationAction: "dlq"
• `$lookup`: include parallelism setting (e.g., parallelism: 2) for concurrent I/O
• AWS connections (S3, Kinesis, Lambda): IAM role ARN must be registered via Atlas Cloud Provider Access first. Always confirm this with user. See references/connection-configs.md for details.

See references/pipeline-patterns.md for stage field examples with JSON syntax.

SchemaRegistry connection: connectionType must be "SchemaRegistry" (not "Kafka"). Schema type values are case-sensitive (use lowercase avro, not AVRO). See references/connection-configs.md for required fields and auth types.

MCP Tool Behaviors

Elicitation: When creating connections, the build tool auto-collects missing sensitive fields (passwords, bootstrap servers) via MCP elicitation. Do NOT ask the user for these — let the tool collect them.

Auto-normalization:

• bootstrapServers array → auto-converted to comma-separated string
• schemaRegistryUrls string → auto-wrapped in array
• dbRoleToExecute → defaults to {role: "readWriteAnyDatabase", type: "BUILT_IN"} for Cluster connections

Workspace creation: includeSampleData defaults to true, which auto-creates the sample_stream_solar connection.

Region naming: The region field uses Atlas-specific names that differ by cloud provider. Using the wrong format returns a cryptic dataProcessRegion error.

See references/connection-configs.md for the full region mapping table. If unsure, inspect an existing workspace with atlas-streams-discover → inspect-workspace and check dataProcessRegion.region.

Connection Capabilities — Source/Sink Reference

Know what each connection type can do before creating pipelines:

Common connection usage mistakes to avoid:

• ❌ Using $externalFunction as sink with execution: "sync" → Must use execution: "async" for sink stage
• ❌ Forgetting change streams exist → Atlas Cluster is a powerful source, not just a sink
• ❌ Using $merge with Kafka → Use $emit for Kafka sinks

See references/connection-configs.md for detailed connection configuration schemas by type.

Core Workflows

Setup from scratch

1. atlas-streams-discover → list-workspaces (check existing) 2. atlas-streams-build → resource: "workspace" (region near data, SP10 for dev) 3. atlas-streams-build → resource: "connection" (for each source/sink/enrichment) 4. Validate connections: atlas-streams-discover → list-connections + inspect-connection for each — verify names match targets, present summary to user 5. Call search-knowledge to validate field names. Fetch relevant examples from https://github.com/mongodb/ASP_example 6. atlas-streams-build → resource: "processor" (with DLQ configured) 7. atlas-streams-manage → start-processor (warn about billing)

Workflow Patterns

Incremental pipeline development (recommended): See references/development-workflow.md for the full 5-phase lifecycle. 1. Start with basic $source → $merge pipeline (validate connectivity) 2. Add $match stages (validate filtering) 3. Add $addFields / $project transforms (validate reshaping) 4. Add windowing or enrichment (validate aggregation logic) 5. Add error handling / DLQ configuration

Modify a processor pipeline: 1. atlas-streams-manage → action: "stop-processor" — processor MUST be stopped first 2. atlas-streams-manage → action: "modify-processor" — provide new pipeline 3. atlas-streams-manage → action: "start-processor" — restart

Debug a failing processor: 1. atlas-streams-discover → diagnose-processor — one-shot health report. Always call this first. 2. Commit to a specific root cause. Match symptoms to diagnostic patterns:

• Error 419 + "no partitions found" → Kafka topic doesn't exist or is misspelled
• State: FAILED + multiple restarts → connection-level error (bypasses DLQ), check connection config
• State: STARTED + zero output + windowed pipeline → likely idle Kafka partitions blocking window closure; add partitionIdleTimeout to Kafka $source (e.g., {"size": 30, "unit": "second"})
• State: STARTED + zero output + non-windowed → check if source has data; inspect Kafka offset lag
• High memoryUsageBytes approaching tier limit → OOM risk; recommend higher tier
• DLQ count increasing → per-document errors; use MongoDB find on DLQ collection

See references/output-diagnostics.md for the full pattern table. 3. Classify processor type before interpreting output volume (alert vs transformation vs filter). 4. Provide concrete, ordered fix steps specific to the diagnosed root cause. Do NOT present a list of hypothetical scenarios. 5. If detailed logs are needed, direct the user to the Atlas UI: Atlas → Stream Processing → Workspace → Processor → Logs tab.

Chained processors (multi-sink pattern)

CRITICAL: A single pipeline can only have ONE terminal sink ($merge or $emit). When users request multiple output destinations (e.g., "write to Atlas AND emit to Kafka"), you MUST acknowledge the single-sink constraint and propose chained processors using an intermediate destination. See references/pipeline-patterns.md for the full pattern with examples.

Pre-Deploy & Post-Deploy Checklists

See references/development-workflow.md for the complete pre-deploy quality checklist (connection validation, pipeline validation) and post-deploy verification workflow.

Tier Sizing & Performance

See references/sizing-and-parallelism.md for tier specifications, parallelism formulas, complexity scoring, and performance optimization strategies.

Troubleshooting

See references/development-workflow.md for the complete troubleshooting table covering processor failures, API errors, configuration issues, and performance problems.

Billing & Cost

Atlas Stream Processing has no free tier. All deployed processors incur continuous charges while running.

• Charges are per-hour, calculated per-second, only while the processor is running
• stop-processor stops billing; stopped processors retain state for 45 days at no charge
• For prototyping without billing: Use sp.process() in mongosh — runs pipelines ephemerally without deploying a processor
• See references/sizing-and-parallelism.md for tier pricing and cost optimization strategies

Safety Rules

• atlas-streams-teardown and atlas-streams-manage require user confirmation — do not bypass
• BEFORE calling `atlas-streams-teardown` for a workspace, you MUST first inspect the workspace with atlas-streams-discover to count connections and processors, then present this information to the user before requesting confirmation
• BEFORE creating any processor, you MUST validate all connections per the "Pre-Deployment Validation" section in references/development-workflow.md
• Deleting a workspace removes ALL connections and processors permanently
• After stopping a processor, state is preserved 45 days — then checkpoints are discarded
• resumeFromCheckpoint: false drops all window state — warn user first
• Moving processors between workspaces is not supported (must recreate)
• Dry-run / simulation is not supported — explain what you would do and ask for confirmation
• Always warn users about billing before starting processors
• Store API authentication credentials in connection settings, never hardcode in processor pipelines

Reference Files

Connection Configuration Reference

Official examples repo: https://github.com/mongodb/ASP_example — check quickstarts, example processors, and Terraform examples. Start with quickstarts.

Connection Capabilities — Source/Sink Reference

Know what each connection type can do before creating pipelines:

Common connection usage mistakes to avoid:

• ❌ Using HTTPS connections as $source → HTTPS is for enrichment or sink only
• ❌ Using $externalFunction as sink with execution: "sync" → Must use execution: "async" for sink stage
• ❌ Forgetting change streams exist → Atlas Cluster is a powerful source, not just a sink
• ❌ Using $merge with Kafka → Use $emit for Kafka sinks

$externalFunction execution modes:

• Mid-pipeline: Can use execution: "sync" (blocks until Lambda returns) or execution: "async" (non-blocking)
• Final sink stage: MUST use execution: "async" only

Connection Naming Best Practices

CRITICAL: Connection names should clearly indicate their actual targets to avoid confusion and prevent writing data to wrong destinations.

Good Naming Patterns

Match the actual target name:

• Cluster connection to "ClusterRestoreTest" → name it cluster-restore-test or ClusterRestoreTest
• Cluster connection to "AtlasCluster" → name it atlas-cluster or AtlasCluster

Use descriptive names with context:

• prod-kafka-orders (indicates environment + service + purpose)
• dev-atlas-main (indicates environment + service + designation)
• staging-s3-exports (indicates environment + service + purpose)

Bad Naming Patterns (AVOID)

❌ Generic names that don't match targets:

• Connection "atlascluster" pointing to "ClusterRestoreTest" ← CONFUSING!
• Connection "kafka" pointing to multiple different topics ← NOT SPECIFIC!

❌ Reusing names across workspaces without context:

• "myconnection" in workspace A and workspace B with different targets

❌ Names that don't indicate connection type:

• "connection1", "test", "temp" ← NO CONTEXT!

Verification Workflow

Before creating processors, always inspect your connections to verify they point where you expect:

1. atlas-streams-discover → action: "list-connections"
2. atlas-streams-discover → action: "inspect-connection" for each
3. Verify connection name matches actual target (clusterName, bootstrapServers, url, etc.)
4. If mismatch exists, consider renaming or warn the user

See development-workflow.md "Pre-Deployment Connection Validation" section for the complete validation procedure.

Important Notes

• HTTPS connections are for $https enrichment ONLY — they are NOT valid as $source data sources
• Store API authentication in connection settings, never hardcode in processor pipelines
• AWS connections (S3, Kinesis, Lambda) require IAM role ARN registered via Atlas Cloud Provider Access first
• Supported connectionType values: Kafka, Cluster, S3, Https, AWSKinesisDataStreams, AWSLambda, SchemaRegistry, Sample

AWS Cloud Provider Access Prerequisites

For S3, Kinesis, and Lambda connections:

AWS connections (S3, Kinesis, Lambda) require that the IAM role ARN be registered in the Atlas project via Cloud Provider Access before creating the connection. This is a prerequisite — the connection creation will fail without it.

Always mention this prerequisite in your response when the user wants to create AWS connections, even if the user says connections already exist. Confirm with language like:

• "IAM role ARNs are registered via Atlas Cloud Provider Access"
• "Ensure IAM role ARNs are registered via Atlas Cloud Provider Access before creating connections"

Security best practice: Use a dedicated IAM role per processor (or group of related processors) with least-privilege permissions scoped only to the specific S3 buckets, Kinesis streams, or Lambda functions that processor needs. Avoid sharing broad-access roles across unrelated processors.

Region Mapping Reference

The region field for workspace creation uses Atlas-specific names that differ by cloud provider. Using the wrong format returns a cryptic dataProcessRegion error.

This is a partial list. If unsure, inspect an existing workspace with atlas-streams-discover → inspect-workspace and check dataProcessRegion.region.

MCP Tool Behaviors for Connections

Elicitation: When required fields are missing, the build tool auto-prompts for them via an interactive form (MCP elicitation protocol). Do NOT manually ask the user for passwords or bootstrap servers — let the tool collect them.

Auto-normalization:

• bootstrapServers passed as array → auto-converted to comma-separated string
• schemaRegistryUrls passed as string → auto-wrapped in array
• Cluster dbRoleToExecute → auto-defaults to {role: "readWriteAnyDatabase", type: "BUILT_IN"} if omitted

connectionConfig by type

Kafka

{
  "bootstrapServers": "broker1:9092,broker2:9092",
  "authentication": {
    "mechanism": "SCRAM-256",
    "username": "my-user",
    "password": "my-password"
  },
  "security": {
    "protocol": "SASL_SSL"
  }
}

Important: bootstrapServers is a comma-separated string, not an array.

All fields above are required. The tool will prompt the user for username/password via elicitation if not provided.

Authentication mechanisms: PLAIN, SCRAM-256, SCRAM-512, OAUTHBEARER Security protocols: SASL_SSL, SASL_PLAINTEXT, SSL

For Confluent Cloud, use mechanism: "PLAIN" with your API key as username and API secret as password.

Kafka supports both PrivateLink and VPC Peering for private networking. See the PrivateLink Reference section below for all supported vendors and providers.

VPC Peering:

• Supported for outbound connections to Kafka brokers in your own VPC
• Requires SASL_SSL security protocol
• Use atlas-streams-manage with accept-peering action to complete the peering setup
• Requires AWS account ID, VPC ID, and region information

Important: Networking cannot be modified after connection creation. To add or change PrivateLink/VPC peering on an existing Kafka connection, you must delete it and recreate it with the networking config.

Use atlas-streams-discover → action: "get-networking" to list available PrivateLink endpoints and VPC peering connections.

Cluster (Atlas)

{
  "clusterName": "my-atlas-cluster",
  "dbRoleToExecute": {
    "role": "readWriteAnyDatabase",
    "type": "BUILT_IN"
  }
}

clusterName is required — must be a cluster in the same project (use atlas-list-clusters to verify).

dbRoleToExecute defaults to {role: "readWriteAnyDatabase", type: "BUILT_IN"} if not provided.

Optional: clusterGroupId (if cluster is in a different project — requires cross-project access to be enabled at the org level).

S3

{
  "aws": {
    "roleArn": "arn:aws:iam::123456789:role/streams-s3-role",
    "testBucket": "my-test-bucket"
  }
}

Prerequisite: The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.

Required IAM policy permissions: s3:ListBucket, s3:GetObject, s3:PutObject.

Https

{
  "url": "https://api.example.com/webhook",
  "headers": {
    "Authorization": "Bearer token123"
  }
}

IMPORTANT: HTTPS connections are for $https enrichment stages ONLY. They are NOT valid data sources — do not use them in $source.

Store all API authentication in the connection config headers, not in the processor pipeline.

HTTPS Auth Patterns

API Key:

{"url": "https://api.example.com", "headers": {"X-API-Key": "your-api-key"}}

Bearer Token:

{"url": "https://api.example.com", "headers": {"Authorization": "Bearer your-token"}}

Basic Auth:

{"url": "https://api.example.com", "headers": {"Authorization": "Basic base64-encoded-credentials"}}

OAuth 2.0 (pre-obtained token):

{"url": "https://api.example.com", "headers": {"Authorization": "Bearer oauth-access-token"}}

AWSKinesisDataStreams

{
  "aws": {
    "roleArn": "arn:aws:iam::123456789:role/streams-kinesis-role"
  }
}

Prerequisite: The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.

Required IAM policy permissions: kinesis:ListShards, kinesis:SubscribeToShard, kinesis:PutRecords, kinesis:DescribeStreamSummary.

AWSLambda

{
  "aws": {
    "roleArn": "arn:aws:iam::123456789:role/streams-lambda-role"
  }
}

Prerequisite: The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.

SchemaRegistry

{
  "connectionType": "SchemaRegistry",
  "connectionConfig": {
    "schemaRegistryUrls": ["https://schema-registry.example.com"],
    "schemaRegistryAuthentication": {
      "type": "USER_INFO",
      "username": "...",
      "password": "..."
    }
  }
}

• connectionType MUST be "SchemaRegistry" (not "Kafka" or "Https")
• schemaRegistryUrls is an array (not a string). The tool auto-wraps a string into an array if needed.
• schemaRegistryAuthentication.type: "USER_INFO" (explicit credentials) or "SASL_INHERIT" (inherit from Kafka connection)
• Tool elicitation will collect sensitive fields (password) — don't ask the user for these directly

Sample

No connectionConfig required. Provides built-in test data. Useful for development and testing without external infrastructure.

Available sample formats: sample_stream_solar (default, auto-created when includeSampleData: true on workspace), samplestock, sampleweather, sampleiot, samplelog, samplecommerce.

PrivateLink Reference (All Vendors)

PrivateLink is supported for Kafka, S3, Kinesis, and Azure EventHub connections. Create a project-level PrivateLink first, then reference it in the connection's networking.access config.

Step 1: Create project-level PrivateLink via atlas-streams-build resource='privatelink':

Step 2: Reference in connection networking config:

{
  "networking": {
    "access": {
      "type": "PRIVATE_LINK",
      "connectionId": "<PrivateLink _id from Step 1>"
    }
  }
}

Use atlas-streams-discover action='get-networking' to find the PrivateLink _id.

Note: Networking config cannot be modified after connection creation — delete and recreate to change.

Development Workflow Reference

Pipeline Stage Categories

Understanding stage categories helps compose valid pipelines. Stages must appear in this order:

Key ordering principle: Place $match as early as possible (reduces volume for all downstream stages). Place $project after $match (reduces document size). Place $https after windows (batches API calls).

5-Phase Development Lifecycle

Phase 1: Project Setup

Goal: Workspace and connections ready.

1. Discover existing resources:

• atlas-streams-discover → list-workspaces — see what already exists
• If workspace exists, inspect-workspace to review config

2. Create workspace (if needed):

• atlas-streams-build → resource: "workspace"
• Choose region close to your data sources
• Start with tier: "SP10" for development
• includeSampleData: true (default) gives you sample_stream_solar for testing

3. Verify workspace:

• atlas-streams-discover → inspect-workspace — confirm state and region

Phase 2: Connection Development

Goal: All data sources and sinks connected and verified.

1. Identify required connections:

• Source connections (Kafka, Cluster change streams, Kinesis, Sample)
• Sink connections (Cluster for $merge, Kafka for $emit, S3, Kinesis)
• Enrichment connections (Https for $https, Cluster for $lookup)

2. Create each connection:

• atlas-streams-build → resource: "connection" for each
• Let the tool elicit missing sensitive fields (passwords, bootstrap servers)
• See connection-configs.md for type-specific schemas

3. Verify connections:

• atlas-streams-discover → list-connections — confirm all created
• atlas-streams-discover → inspect-connection for each — verify state and config

Phase 3: Processor Development

Goal: Working processor with validated pipeline.

Pre-Deployment Connection Validation (MANDATORY)

BEFORE creating any processor, you MUST validate all connections referenced in your pipeline. This prevents silent failures and confusion about data destinations.

Step 1: List all connections in workspace

atlas-streams-discover → action: "list-connections", workspaceName: "<your-workspace>"

Verify all required connections exist.

Step 2: Inspect EACH connection referenced in pipeline

For EVERY connectionName in your pipeline (source, sink, enrichment), inspect it:

atlas-streams-discover → action: "inspect-connection",
                         workspaceName: "<your-workspace>",
                         resourceName: "<connection-name>"

Verify for each connection:

• [ ] Connection exists and state is READY
• [ ] Connection type matches intended usage:
• Cluster: valid for $source (change streams), $merge, $lookup
• Kafka: valid for $source, $emit
• S3: valid for $emit only
• Https: valid for $https enrichment or sink
• Lambda: valid for $externalFunction only
• [ ] Connection name matches actual target (avoid confusion):
• ⚠️ BAD: connection "atlascluster" → actual target "ClusterRestoreTest"
• ✅ GOOD: connection "cluster-restore-test" → actual target "ClusterRestoreTest"
• [ ] For Cluster connections: verify the clusterName field points to the intended cluster

Step 3: Present validation summary to user

Always show the user what connections will be used:

"Before creating processor '<name>', I've verified your connections:
 - ✅ sample_stream_solar → Sample data (READY)
 - ⚠️ atlascluster → ClusterRestoreTest (READY)
      Warning: Connection name 'atlascluster' doesn't match actual cluster 'ClusterRestoreTest'
 - ✅ open-meteo-api → https://api.open-meteo.com/v1/... (READY)

Proceed with processor creation?"

Step 4: Wait for user confirmation if warnings exist

If any connection name doesn't match its target, ask the user to confirm before proceeding.

Step 5: Only then create the processor

This validation workflow prevents:

• Creating processors with non-existent connections (fails immediately)
• Writing data to unexpected clusters (e.g., "atlascluster" → "ClusterRestoreTest" instead of "AtlasCluster")
• Confusion when verifying output data later

Incremental Pipeline Development

Follow incremental pipeline development — test at each step:

Step 1: Basic connectivity

[
  {"$source": {"connectionName": "my-source"}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step1"}}}
]

Create with autoStart: true. Verify documents flow. Stop processor.

Step 2: Add filtering

[
  {"$source": {"connectionName": "my-source"}},
  {"$match": {"status": "active"}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step2"}}}
]

Modify pipeline (stop → modify-processor → start). Verify filtered output.

Step 3: Add transformations

[
  {"$source": {"connectionName": "my-source"}},
  {"$match": {"status": "active"}},
  {"$addFields": {"processed_at": "$$NOW_NOT_VALID"}},
  {"$project": {"userId": 1, "amount": 1, "processed_at": 1}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step3"}}}
]

Remember: $$NOW is NOT valid in streaming. Use a field from the source document or omit.

Step 4: Add windowing or enrichment (if needed)

Step 5: Add error handling

• Configure DLQ: {"dlq": {"connectionName": "my-sink", "db": "streams_dlq", "coll": "failed_docs"}}
• Add $ifNull for optional enrichment fields
• Set onError: "dlq" on $https stages

Phase 4: Testing & Validation

Goal: Processor verified working correctly.

1. Confirm processor state:

• atlas-streams-discover → inspect-processor — state should be STARTED

2. Run diagnostics:

• atlas-streams-discover → diagnose-processor — full health report

3. Verify data flow:

• Use MongoDB count tool on output collection — documents arriving?
• Use MongoDB find tool on output collection — data looks correct?
• Use MongoDB count tool on DLQ collection — any errors?
• If DLQ has documents, use MongoDB find tool to inspect failure reasons

4. Classify output volume:

• See output-diagnostics.md for the full decision framework
• Alert processors: low output is expected
• Transformation processors: low output is a red flag

Phase 5: Production Deployment

Goal: Processor running at appropriate tier with monitoring.

1. Right-size the tier:

• See sizing-and-parallelism.md for tier selection
• Review memoryUsageBytes from diagnostics
• Consider parallelism needs for $merge, $lookup, $https
• Upgrade tier: atlas-streams-manage → stop-processor, then start-processor with tier override

2. Ensure DLQ is configured (mandatory for production)

3. Use descriptive processor names (e.g., fraud-detector, order-enricher, iot-rollup)

Debugging Decision Trees

Connection Failures

1. atlas-streams-discover → inspect-connection — check state 2. If Kafka: verify bootstrapServers is a comma-separated string (not array) 3. If Cluster: verify cluster exists in project (atlas-list-clusters) 4. If AWS (S3/Kinesis/Lambda): verify IAM role ARN is registered in Cloud Provider Access 5. If Https: verify URL is reachable and auth headers are in connection config

Processor Startup Failures

1. atlas-streams-discover → diagnose-processor — check state and errors 2. If FAILED: read the error message in diagnostics 3. Common causes:

• Invalid pipeline syntax (missing $source, missing sink)
• $$NOW/$$ROOT/$$CURRENT used (not valid in streaming)
• Kafka $source missing topic field
• Referenced connection doesn't exist — validate with list-connections first
• Connection name doesn't match expected target — inspect connection to verify actual cluster/resource
• OOM — tier too small for pipeline complexity

Processing Errors (Running but DLQ filling up)

1. Use MongoDB find tool on DLQ collection — inspect error messages 2. Common causes:

• Schema mismatches in source data
• $https enrichment failures (API down, auth expired)
• Type errors in $addFields or $project expressions

3. Fix: stop-processor → modify-processor (fix pipeline) → start-processor

Performance Issues (Running but slow)

1. atlas-streams-discover → diagnose-processor — check stats 2. Check memoryUsageBytes — if near 80% of tier RAM, upgrade tier 3. Check if $match is early in pipeline (reduces downstream volume) 4. Check if $https has parallelism setting (increase for I/O-bound enrichment) 5. Check if windows have partitionIdleTimeout (idle Kafka partitions block windows) 6. Consider upgrading tier or increasing stage parallelism

Operational Monitoring Cadence

Daily

• Check processor states via atlas-streams-discover → list-processors
• Verify DLQ collections aren't growing via MongoDB count tool
• Confirm output collections are receiving data

Weekly

• Run diagnose-processor for each production processor
• Review memoryUsageBytes trends — approaching 80%?
• Check connection health across all connections

Monthly

• Evaluate tier appropriateness — over-provisioned or under-provisioned?
• Review DLQ patterns — recurring errors that need pipeline fixes?
• Consider parallelism adjustments based on throughput trends

Troubleshooting

Pre-Deploy Quality Checklist

Before creating a processor, verify:

Connection Validation (MANDATORY - Always do this first)

• [ ] CRITICAL: Call atlas-streams-discover → action: "list-connections" to list all connections in workspace
• [ ] CRITICAL: Call atlas-streams-discover → action: "inspect-connection" for EACH connection referenced in pipeline
• [ ] CRITICAL: Verify connection names clearly indicate their actual targets (avoid generic names like "atlascluster" pointing to "ClusterRestoreTest")
• [ ] CRITICAL: Present connection summary to user: "Connection 'X' → Actual target 'Y'" for each connection
• [ ] CRITICAL: Warn user if connection names don't match their targets and ask for confirmation
• [ ] All connections are in READY state
• [ ] Connection types match usage (Cluster for $source/$merge, Kafka for topics, etc.)

Pipeline Validation

• [ ] search-knowledge was called to validate sink/source field names
• [ ] Pipeline starts with $source and ends with $merge, $emit, $https, or $externalFunction (async)
• [ ] No $$NOW, $$ROOT, or $$CURRENT in the pipeline
• [ ] Kafka $source includes a topic field
• [ ] Kafka $source with windowed pipeline includes partitionIdleTimeout (prevents windows from stalling on idle partitions)
• [ ] HTTPS connections are only used in $https enrichment or sink stages, not in $source
• [ ] DLQ is configured (recommended for production)
• [ ] $https stages use onError: "dlq" (not "fail")
• [ ] $externalFunction stages use onError: "dlq" and execution is explicitly set
• [ ] API auth is stored in connection settings, not hardcoded in the pipeline

Post-Deploy Verification Workflow

After creating and starting a processor: 1. atlas-streams-discover → action: "inspect-processor" — confirm state is STARTED 2. atlas-streams-discover → action: "diagnose-processor" — check for errors in the health report 3. Use MongoDB count tool on the DLQ collection — verify no errors accumulating 4. Use MongoDB find tool on the output collection — verify documents are arriving 5. If output is low/zero, classify processor type before assuming a problem (see Debug section)

MCP Server Troubleshooting

This skill requires the MongoDB MCP Server with Atlas Stream Processing tools enabled. If these tools are unavailable, follow the diagnostic steps below.

Step 1: Verify MCP Server Connection

Check if the MongoDB MCP Server is connected to your environment.

If not connected:

• Install the MongoDB MCP Server
• Configure it with your Atlas API credentials (apiClientId and apiClientSecret)

Step 2: Verify Tool Availability

Check that all four streams tools are available:

• atlas-streams-discover
• atlas-streams-build
• atlas-streams-manage
• atlas-streams-teardown

Fallback Options (Limited Functionality)

If you cannot configure the MCP server immediately, you have limited alternatives:

Option 1: Atlas CLI (Read-Only)

Use Atlas CLI API commands for exploration only:

atlas api streams listStreamWorkspaces --projectId <project-id>
atlas api streams getStreamWorkspace --workspaceName <workspace-name> --projectId <project-id>

Limitations:

• Read-only operations only
• Cannot create or modify processors
• No automated validation or diagnostics

Option 2: mongosh with sp.process() (Prototyping Only)

Use sp.process() in mongosh for ephemeral pipeline testing:

sp.process([
  { $source: { connectionName: "sample_stream_solar" } },
  { $match: { temperature: { $gt: 50 } } },
  { $limit: 10 }
])

Limitations:

• Ephemeral only (no deployed processors)
• No billing (runs locally)
• Cannot test production connections
• Limited to simple pipeline validation

Recommended Action

For full Atlas Stream Processing capabilities, configure the MongoDB MCP Server with streams preview features enabled. The fallback options above provide minimal functionality and are not suitable for production workflows.

Processor Output Diagnostics Reference

The Problem

A user says "my processor isn't outputting anything" or "output seems low." Before assuming something is broken, you must classify the processor type — low output may be perfectly normal.

Processor Type Classification

Category 1: Alert / Anomaly Detection

Expected output: Low or zero most of the time. Spikes during anomalous events.

Examples:

• Fraud detection (flags suspicious transactions)
• Threshold alerting (temperature > 100, latency > 500ms)
• Error monitoring (filters for error-level events)
• Security alerting (unusual login patterns)

Green flags (healthy):

• Zero output during normal conditions
• Occasional bursts during genuine anomalies
• DLQ is empty or near-empty

Red flags (problem):

• Zero output during a known anomaly event
• DLQ filling up with errors
• Processor state is FAILED

Category 2: Data Transformation / Ingestion

Expected output: Roughly 1:1 with input volume. Output should be proportional to source.

Examples:

• Format conversion (Kafka → Atlas)
• Data enrichment (add fields, lookup)
• Schema normalization
• Archive pipelines (collection → collection)

Green flags (healthy):

• Output volume roughly matches input volume
• Consistent throughput over time

Red flags (problem):

• Output is zero while source has data
• Output is much lower than expected source volume
• Growing backlog (source advancing but output not keeping up)
• DLQ accumulating documents

Category 3: Filter / Quality Gate

Expected output: Variable — depends on match rate of filter criteria.

Examples:

• Quality filtering ($match for valid records)
• Data routing (priority-based splitting)
• Deduplication
• Sampling

Green flags (healthy):

• Output is a consistent percentage of input
• Percentage aligns with expected data quality/match rate

Red flags (problem):

• Output drops to zero when source has data
• Sudden change in output ratio without a data source change
• DLQ filling up (filter errors, not just filtered-out data)

Diagnostic Workflow

Step 1: Classify the processor

Ask the user what the processor does, or inspect the pipeline:

• atlas-streams-discover → inspect-processor — read the pipeline stages

Classification heuristics from pipeline:

• Has $match with narrow conditions (e.g., severity > 8) → likely Alert
• Pipeline is mostly $addFields/$project/$merge → likely Transformation
• $match filters broadly (e.g., status: "active") → likely Filter
• Has $tumblingWindow with $match inside → likely Alert (windowed anomaly detection)
• Has $tumblingWindow with $group only → likely Transformation (aggregation)

Step 2: Check processor state

• atlas-streams-discover → diagnose-processor
• If state is FAILED → the problem is not low output, it's a crash. See debugging trees in development-workflow.md.

Step 3: Check operational logs

• For detailed logs, direct the user to the Atlas UI: Atlas → Stream Processing → Workspace → Processor → Logs tab
• Operational logs contain runtime errors: Kafka producer/consumer failures, schema serialization issues, OOM events, connection timeouts

Step 4: Check DLQ

• Use MongoDB count tool on the DLQ collection
• If DLQ has documents → use MongoDB find tool to inspect error messages
• Growing DLQ means documents are being rejected, not that nothing is flowing

Step 5: Check output collection

• Use MongoDB count tool on the output collection
• Use MongoDB find tool with sort: {"_id": -1} and limit: 5 to see most recent documents
• Check timestamps — are documents recent?

Step 6: Interpret based on processor type

Common Diagnostic Patterns

After running diagnose-processor, match the symptoms to these patterns:

When providing fix steps:

• Commit to a specific root cause based on the evidence
• Do NOT present a list of hypothetical scenarios
• Provide concrete, ordered steps (e.g., "stop → modify pipeline to add partitionIdleTimeout → restart with resumeFromCheckpoint: false")

Contextual Factors

Before concluding there's a problem, consider:

• Time of day: Business-hours-only data sources produce nothing at night
• Seasonality: Holiday periods, end-of-month spikes, etc.
• Source health: Is the source (Kafka topic, collection) actually receiving data?
• Window timing: Windowed processors only emit when the window closes — a 5-minute tumbling window outputs nothing for up to 5 minutes after start
• Idle partitions: Kafka windows won't close if a partition has no data — check partitionIdleTimeout

Best Practice: Document Expected Behavior

When creating processors, encourage users to use descriptive names that indicate the processor type:

Pipeline Patterns Reference

Official examples repo: https://github.com/mongodb/ASP_example (quickstarts, example processors, Terraform examples). Start with example_processors/README.md for the full pattern catalog. Always consult the official repo for the latest validated patterns before creating processors.

Stage Quick-Reference

Invalid Constructs

Do NOT use these in streaming pipelines:

• $$NOW, $$ROOT, $$CURRENT — not available in stream processing
• HTTPS connections as $source — HTTPS is for $https enrichment only
• Kafka $source without topic — topic field is required
• Pipelines without a sink — $merge/$emit required for deployed processors (sinkless only works via sp.process())
• Lambda connections with $emit — Lambda uses $externalFunction (can be mid-pipeline or terminal sink with async execution), not $emit

Source Patterns

MongoDB Change Stream

{"$source": {"connectionName": "my-cluster"}}

With full document and pushdown pipeline:

{"$source": {
  "connectionName": "my-cluster",
  "db": "mydb", "coll": "mycoll",
  "fullDocument": "updateLookup",
  "fullDocumentBeforeChange": "whenAvailable",
  "pipeline": [{"$match": {"operationType": "insert"}}]
}}

Kafka (topic is REQUIRED)

{"$source": {
  "connectionName": "my-kafka",
  "topic": "my-topic",
  "auto_offset_reset": "earliest",
  "partitionIdleTimeout": {"size": 30, "unit": "second"}
}}

Kinesis

{"$source": {
  "connectionName": "my-kinesis",
  "stream": "my-stream",
  "config": {"initialPosition": "TRIM_HORIZON"},
  "shardIdleTimeout": {"size": 30, "unit": "second"},
  "consumerARN": "arn:aws:kinesis:us-east-1:123456789:stream/my-stream/consumer/my-consumer:123"
}}

stream (required): Kinesis stream name. config.initialPosition: TRIM_HORIZON (oldest, default) or LATEST. shardIdleTimeout: unblocks windows when shards go idle (like Kafka partitionIdleTimeout). consumerARN (optional): enables enhanced fan-out for dedicated throughput.

Inline Documents (ephemeral testing only)

{"$source": {"documents": [{"device_id": "sensor-1", "temp": 72.5}]}}

Sink Patterns

$merge to Atlas

{"$merge": {"into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"}}}

With match behavior and parallelism:

{"$merge": {
  "into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"},
  "on": "_id", "whenMatched": "replace", "whenNotMatched": "insert",
  "parallelism": 4
}}

whenMatched: replace, merge, delete (via $cond). whenNotMatched: insert.

Additive merge (append to arrays):

{"$merge": {
  "into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"},
  "on": "device_id",
  "whenMatched": [{"$addFields": {"readings": {"$concatArrays": ["$readings", "$$new.readings"]}}}],
  "whenNotMatched": "insert"
}}

Dynamic routing:

{"$merge": {"into": {
  "connectionName": "my-atlas", "db": "mydb",
  "coll": {"$cond": {"if": {"$eq": ["$priority", "high"]}, "then": "alerts", "else": "events"}}
}}}

$emit to Kafka

{"$emit": {
  "connectionName": "my-kafka", "topic": "output-topic",
  "key": {"field": "device_id", "format": "string"}
}}

Key formats: string, json, int, long, binData. Tombstone support: "tombstoneWhen": {"$expr": {"$eq": ["$status", "deleted"]}}.

$emit to Kafka with Schema Registry (Avro)

{"$emit": {
  "connectionName": "my-kafka", "topic": "output-topic",
  "schemaRegistry": {
    "connectionName": "my-schema-registry",
    "valueSchema": {
      "type": "avro",
      "schema": {
        "type": "record", "name": "SensorReading",
        "fields": [
          {"name": "device_id", "type": "string"},
          {"name": "temp", "type": "double"},
          {"name": "timestamp", "type": "long"}
        ]
      },
      "options": {
        "subjectNameStrategy": "TopicNameStrategy",
        "autoRegisterSchemas": true
      }
    }
  }
}}

Requires a SchemaRegistry connection (see connection-configs.md). valueSchema.type must be lowercase avro (case-sensitive). valueSchema.schema is always required, even with autoRegisterSchemas: true.

$emit to Kinesis

{"$emit": {"connectionName": "my-kinesis", "stream": "out", "partitionKey": "$device_id"}}

$emit to S3

{"$emit": {
  "connectionName": "my-s3", "bucket": "my-bucket",
  "path": {"$concat": ["data/", {"$dateToString": {"format": "%Y/%m/%d", "date": "$timestamp"}}]},
  "config": {"outputFormat": "relaxedJson"}
}}

Fields: connectionName (required), bucket (required), path (required — key prefix string or expression), region (optional), config (optional — outputFormat, writeOptions, delimiter, compression).

$https as Sink (webhook/API)

{"$https": {
  "connectionName": "my-webhook",
  "path": "/events",
  "method": "POST",
  "onError": "dlq"
}}

When used as a final sink stage, $https sends processed documents to an external HTTP endpoint. Unlike mid-pipeline usage (which enriches documents with API responses), sink usage doesn't expect a response to merge back into the document. Useful for:

• Sending data to webhooks
• Posting to external APIs
• Triggering external systems

$externalFunction as Sink (Lambda async)

{"$externalFunction": {
  "connectionName": "my-lambda",
  "functionName": "arn:aws:lambda:us-west-1:123456789:function:my-function",
  "execution": "async",
  "onError": "dlq"
}}

Important: When used as a final sink stage, $externalFunction MUST use execution: "async". This fires off the Lambda function without waiting for a response, useful for:

• Triggering downstream AWS applications or analytics
• Notifying external systems
• Firing off alerts or billing logic
• Propagating data to external workflows

Unlike mid-pipeline usage (where execution: "sync" is allowed for enrichment), sink usage requires async execution only. The pipeline still needs this as the terminal stage — you cannot use $emit to invoke Lambda.

Window Patterns

Tumbling

{"$tumblingWindow": {
  "interval": {"size": 5, "unit": "minute"},
  "pipeline": [{"$group": {"_id": "$deviceId", "avg": {"$avg": "$temp"}, "count": {"$sum": 1}}}]
}}

Hopping (with allowedLateness)

{"$hoppingWindow": {
  "interval": {"size": 5, "unit": "minute"},
  "hopSize": {"size": 1, "unit": "minute"},
  "allowedLateness": {"size": 15, "unit": "second"},
  "pipeline": [{"$group": {"_id": "$region", "total": {"$sum": "$amount"}}}]
}}

Session

{"$sessionWindow": {
  "gap": {"size": 5, "unit": "minute"}, "key": "$userId",
  "pipeline": [{"$group": {"_id": "$userId", "actions": {"$push": "$action"}, "count": {"$sum": 1}}}]
}}

Late data

{"$tumblingWindow": {
  "interval": {"size": 1, "unit": "minute"},
  "allowedLateness": {"size": 30, "unit": "second"},
  "boundaryType": "eventTime",
  "pipeline": [{"$group": {"_id": "$sensorId", "max": {"$max": "$value"}}}]
}}

boundaryType: eventTime (document timestamp) or processTime (wall clock, default).

Windowing Rules

• Windows require $group inside the window pipeline
• Idle Kafka partitions block windows — use partitionIdleTimeout
• allowedLateness lets late docs update closed windows

Enrichment Patterns

$https

{"$https": {
  "connectionName": "my-api",
  "path": {"$concat": ["/users/", "$userId"]},
  "method": "GET", "as": "userInfo", "onError": "dlq"
}}

onError: dlq (recommended), discard, fail. Store auth in connection settings, not pipeline. Place $https after windows to batch requests.

$lookup

{"$lookup": {
  "connectionName": "my-atlas",
  "from": {"db": "mydb", "coll": "users"},
  "localField": "userId", "foreignField": "_id", "as": "user",
  "parallelism": 2
}}

$externalFunction (Lambda - Mid-Pipeline Enrichment)

{"$externalFunction": {
  "connectionName": "my-lambda",
  "functionName": "my-function-name",
  "execution": "sync",
  "as": "lambdaResult",
  "onError": "dlq",
  "payload": [
    {"$project": {"userId": 1, "data": 1}}
  ]
}}

Mid-pipeline usage:

• execution: sync (waits for Lambda result, stores in as field) or async (non-blocking)
• as: Field name to store Lambda response (required for sync, ignored for async)
• payload: Optional inner pipeline to customize request body sent to Lambda
• Use for enriching/transforming documents before downstream stages

Sink usage: See the Sink Patterns section. When used as final stage, MUST use execution: "async" only.

$validate (Schema Validation)

{"$validate": {
  "validator": {"$jsonSchema": {
    "required": ["device_id", "timestamp", "reading"],
    "properties": {
      "device_id": {"bsonType": "string"},
      "reading": {"bsonType": "double"}
    }
  }},
  "validationAction": "dlq"
}}

validationAction: "dlq" (recommended), "discard", "error" (crashes processor — avoid in production). Place early to catch bad data before expensive stages.

$function (JavaScript UDF)

{"$addFields": {
  "boostedWatts": {"$function": {
    "body": "function(watts) { return watts * 1.2; }",
    "args": ["$watts"],
    "lang": "js"
  }}
}}

Requires SP30+ tier. body: JavaScript function as string. args: array of field references. lang: always "js".

Common Pipeline Patterns

Array Normalization

[
  {"$source": {"connectionName": "my-kafka", "topic": "orders"}},
  {"$unwind": "$items"},
  {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$items", {"orderId": "$orderId", "ts": "$timestamp"}]}}},
  {"$merge": {"into": {"connectionName": "my-atlas", "db": "mydb", "coll": "line_items"}}}
]

Dynamic Kafka Topic Routing

{"$emit": {
  "connectionName": "my-kafka",
  "topic": {"$switch": {
    "branches": [
      {"case": {"$eq": ["$severity", "critical"]}, "then": "alerts-critical"},
      {"case": {"$eq": ["$severity", "warning"]}, "then": "alerts-warning"}
    ],
    "default": "alerts-info"
  }}
}}

Complex Event Processing (Fraud Detection)

[
  {"$source": {"connectionName": "my-kafka", "topic": "transactions"}},
  {"$tumblingWindow": {
    "interval": {"size": 5, "unit": "minute"},
    "pipeline": [
      {"$group": {
        "_id": "$userId",
        "txnCount": {"$sum": 1},
        "totalAmount": {"$sum": "$amount"},
        "uniqueLocations": {"$addToSet": "$location"}
      }},
      {"$addFields": {
        "suspiciousLocations": {"$gt": [{"$size": "$uniqueLocations"}, 3]},
        "highVelocity": {"$gt": ["$txnCount", 10]}
      }},
      {"$match": {"$or": [{"suspiciousLocations": true}, {"highVelocity": true}]}}
    ]
  }},
  {"$merge": {"into": {"connectionName": "my-atlas", "db": "fraud", "coll": "alerts"}}}
]

Graceful Degradation with $ifNull

{"$addFields": {
  "userName": {"$ifNull": ["$userInfo.name", "unknown"]},
  "userTier": {"$ifNull": ["$userInfo.tier", "standard"]},
  "enrichmentSucceeded": {"$ne": [{"$type": "$userInfo"}, "missing"]}
}}

Window Metadata

Inside window pipelines, _stream_meta.window.start and _stream_meta.window.end provide boundary timestamps:

{"$group": {
  "_id": "$deviceId",
  "windowStart": {"$first": "$_stream_meta.window.start"},
  "windowEnd": {"$first": "$_stream_meta.window.end"},
  "avg": {"$avg": "$temp"}
}}

Checkpoint Resume Constraints

With resumeFromCheckpoint: true (default), you CANNOT change: window type, interval, remove windows, or modify $source. Set false to make these changes (restarts from beginning).

DLQ Configuration

{"dlq": {"connectionName": "my-atlas", "db": "streams_dlq", "coll": "failed_documents"}}

DLQ documents include: original document, error message, stage info, timestamp.

Sample Stream Formats

Chained Processors (Multi-Sink Pattern)

CRITICAL: A single pipeline can only have ONE terminal sink ($merge or $emit). You CANNOT have both $merge and $emit as terminal stages. When a user requests multiple output destinations (e.g., "write to Atlas AND emit to Kafka" or "archive to S3 AND send to Lambda"), you MUST:

1. Acknowledge the single-sink constraint explicitly in your response 2. Propose chained processors: Processor A reads source → enriches → writes to intermediate via $merge (Atlas) or $emit (Kafka). Processor B reads from that intermediate (change stream or Kafka topic) → emits to second destination. Kafka-as-intermediate is lower latency; Atlas-as-intermediate is simpler to inspect. 3. Show both processor pipelines including any $lookup enrichment stages with parallelism settings.

Note: $externalFunction (Lambda) can be used mid-pipeline OR as a terminal sink (with execution: "async"). A pipeline with mid-pipeline $externalFunction AND a terminal $merge/$emit is a valid single-sink pattern (Lambda enriches, then the result is written to the sink).

Required Field Examples by Stage

$source (Kinesis)

Use stream (NOT streamName or topic) for the Kinesis stream name.

{"$source": {"connectionName": "my-kinesis", "stream": "my-stream"}}

$source (change stream)

Include fullDocument: "updateLookup" to get the full document content.

$emit (Kinesis)

MUST include partitionKey.

{"$emit": {"connectionName": "my-kinesis", "stream": "my-stream", "partitionKey": "$fieldName"}}

$emit (S3)

Use path (NOT prefix).

{"$emit": {"connectionName": "my-s3", "bucket": "my-bucket", "path": "data/year={$year}", "config": {"outputFormat": {"name": "json"}}}}

Sizing & Parallelism Reference

Tier Hardware Specs

Memory rule: 20% is reserved for overhead. User state (window accumulation, sort buffers) must stay below 80% of tier RAM. Exceeding this causes OOM failure.

How Parallelism Works

Every stage in a pipeline runs with default parallelism: 1. This base level is included in your tier at no additional cost.

When you need higher throughput for specific stages, increase their parallelism beyond 1. Only values > 1 count toward your tier's maximum.

Stages that commonly benefit from parallelism:

• $merge — concurrent writes to Atlas
• $lookup — concurrent reads for enrichment
• $https — concurrent API calls

Parallelism Calculation

Formula: Total Parallelism = sum of (parallelism - 1) for all stages where parallelism > 1

Tier Selection Algorithm

If Total Parallelism = 0:   → SP2  (max 1)
If Total Parallelism = 1:   → SP5  (max 2)
If Total Parallelism ≤ 8:   → SP10 (max 8)
If Total Parallelism ≤ 16:  → SP30 (max 16)
If Total Parallelism ≤ 64:  → SP50 (max 64)

Worked Examples

Simple pipeline (all parallelism = 1):

$source:    parallelism = 1  (does not count)
$match:     parallelism = 1  (does not count)
$merge:     parallelism = 1  (does not count)

Total = 0 → SP2

Medium pipeline:

$source:    parallelism = 1  (does not count)
$match:     parallelism = 1  (does not count)
$lookup:    parallelism = 4  (counts as 3)
$merge:     parallelism = 4  (counts as 3)

Total = 3 + 3 = 6 → SP10 (max 8)

Complex pipeline:

$source:    parallelism = 1  (does not count)
$https:     parallelism = 6  (counts as 5)
$merge:     parallelism = 8  (counts as 7)

Total = 5 + 7 = 12 → SP30 (max 16)

API Error for Parallelism Exceeded

If you specify a tier too small for the pipeline's parallelism, the API returns:

"Operator parallelism requested exceeds limit for this tier.
(Requested: X, Limit: Y). Minimum tier for this workload: SPxx or larger."

Solution: Use atlas-streams-manage → stop-processor, then start-processor with a higher tier value.

Complexity-Based Tier Selection

When parallelism is all default (1), choose tier based on pipeline complexity:

Complexity Scoring Heuristic

For automated tier recommendation, score the pipeline:

Score → Tier mapping:

• 0-10: SP2
• 11-20: SP5
• 21-40: SP10
• 41-60: SP30
• 61+: SP50

Always take the higher of complexity-driven vs parallelism-driven tier recommendations.

Billing

Charges are per-hour, calculated per-second, only while the processor is running.

• start-processor begins billing
• stop-processor stops billing
• Stopped processors retain state for 45 days at no charge

What's included in the tier price:

• Compute (vCPU and RAM)
• State storage
• Base parallelism (parallelism = 1 for all stages)

Additional costs (separate from tier):

• Data transfer egress (varies by cloud provider and transfer type: intra-region, inter-region, internet)
• VPC Peering (AWS and GCP)
• Private Link connectivity

For current pricing: https://www.mongodb.com/docs/atlas/billing/stream-processing-costs/

Sizing Workflow with MCP Tools

Phase 1: Pre-deployment estimate

1. Score the pipeline using the complexity heuristic above 2. Calculate parallelism needs using the formula 3. Take the higher recommendation 4. Start with that tier (or one tier lower for cost savings during testing)

Phase 2: Validation

1. Deploy the processor: atlas-streams-build → resource: "processor" with autoStart: true 2. Let it run for a representative period 3. Check stats: atlas-streams-discover → diagnose-processor 4. Review memoryUsageBytes:

• Below 50% of tier RAM → over-provisioned, consider downsizing
• 50-70% → good fit
• 70-80% → at limit, monitor closely
• Above 80% → under-provisioned, upgrade before it OOMs

Phase 3: Optimization

1. Stop processor: atlas-streams-manage → stop-processor 2. Restart with adjusted tier: atlas-streams-manage → start-processor with tier override 3. Monitor for another period 4. Repeat until right-sized

Cost Optimization: Time-of-Day Strategy

For workloads with predictable traffic patterns, adjust tiers by time of day:

To change tiers: stop-processor → start-processor with new tier value. Note: resumeFromCheckpoint: true (default) preserves state across tier changes.