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.