feat: add Atlas Stream Processing tools MCP-406

.github/workflows/code-health-fork.yml

@@ -38,7 +38,7 @@ jobs:
       - name: Run tests
         run: pnpm test
         env:
-          SKIP_ATLAS_TESTS: "true"
+          SKIP_ATLAS_INTEGRATION_TESTS: "true"
           SKIP_ATLAS_LOCAL_TESTS: "true"
   run-atlas-local-tests:
     name: Run Atlas Local tests

.github/workflows/code-health.yml

@@ -39,7 +39,7 @@ jobs:
       - name: Run tests
         run: pnpm test
         env:
-          SKIP_ATLAS_TESTS: "true"
+          SKIP_ATLAS_INTEGRATION_TESTS: "true"
           SKIP_ATLAS_LOCAL_TESTS: "true"
           MDB_MONGOT_PASSWORD: ${{ secrets.TEST_MDB_MONGOT_PASSWORD }}
           MDB_VOYAGE_API_KEY: ${{ secrets.TEST_MDB_MCP_VOYAGE_API_KEY }}

package.json

@@ -91,7 +91,7 @@
     "test:accuracy": "sh ./scripts/accuracy/runAccuracyTests.sh",
     "test:browser": "pnpm --filter browser test",
     "test:long-running-tests": "vitest --project long-running-tests --coverage",
-    "test:local": "SKIP_ATLAS_TESTS=true SKIP_ATLAS_LOCAL_TESTS=true pnpm run test",
+    "test:local": "SKIP_ATLAS_INTEGRATION_TESTS=true SKIP_ATLAS_LOCAL_TESTS=true pnpm run test",
     "atlas:cleanup": "vitest --project atlas-cleanup"
   },
   "license": "Apache-2.0",

scripts/cleanupAtlasTestLeftovers.test.ts

@@ -4,6 +4,10 @@ import { ConsoleLogger } from "../src/common/logging/index.js";
 import { Keychain } from "../src/lib.js";
 import { describe, it } from "vitest";
 
+function sleep(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
 function isOlderThanTwoHours(date: string): boolean {
     const twoHoursInMs = 2 * 60 * 60 * 1000;
     const projectDate = new Date(date);
@@ -35,6 +39,78 @@ async function findAllTestProjects(client: ApiClient, orgId: string): Promise<Gr
     return testProjects.filter((proj) => isOlderThanTwoHours(proj.created));
 }
 
+async function deleteAllWorkspacesOnStaleProject(client: ApiClient, projectId: string): Promise<string[]> {
+    const errors: string[] = [];
+
+    try {
+        const workspaces = await client
+            .listStreamWorkspaces({
+                params: {
+                    path: {
+                        groupId: projectId,
+                    },
+                },
+            })
+            .then((res) => res.results || []);
+
+        await Promise.allSettled(
+            workspaces.map(async (workspace) => {
+                const name = workspace.name || "";
+                try {
+                    // Delete all processors first (auto-stops running ones)
+                    try {
+                        const processors = await client
+                            .getStreamProcessors({
+                                params: { path: { groupId: projectId, tenantName: name } },
+                            })
+                            .then((res) => res.results || []);
+                        await Promise.allSettled(
+                            processors.map((p) =>
+                                client.deleteStreamProcessor({
+                                    params: {
+                                        path: {
+                                            groupId: projectId,
+                                            tenantName: name,
+                                            processorName: p.name || "",
+                                        },
+                                    },
+                                })
+                            )
+                        );
+                    } catch {
+                        // Ignore errors listing/deleting processors
+                    }
+                    await client.deleteStreamWorkspace({
+                        params: {
+                            path: { groupId: projectId, tenantName: name },
+                        },
+                    });
+                    // Wait for workspace to be fully deleted (up to 120s)
+                    for (let i = 0; i < 120; i++) {
+                        try {
+                            await client.getStreamWorkspace({
+                                params: {
+                                    path: { groupId: projectId, tenantName: name },
+                                },
+                            });
+                            await sleep(1000);
+                        } catch {
+                            break;
+                        }
+                    }
+                    console.log(`  Deleted workspace: ${name}`);
+                } catch (error) {
+                    errors.push(`Failed to delete workspace ${name} in project ${projectId}: ${String(error)}`);
+                }
+            })
+        );
+    } catch {
+        // Project may not have streams enabled, ignore
+    }
+
+    return errors;
+}
+
 async function deleteAllClustersOnStaleProject(client: ApiClient, projectId: string): Promise<string[]> {
     const errors: string[] = [];
 
@@ -50,12 +126,13 @@ async function deleteAllClustersOnStaleProject(client: ApiClient, projectId: str
 
     await Promise.allSettled(
         allClusters.map(async (cluster) => {
+            const name = cluster.name || "";
             try {
                 await client.deleteCluster({
-                    params: { path: { groupId: projectId || "", clusterName: cluster.name || "" } },
+                    params: { path: { groupId: projectId || "", clusterName: name } },
                 });
             } catch (error) {
-                errors.push(`Failed to delete cluster ${cluster.name} in project ${projectId}: ${String(error)}`);
+                errors.push(`Failed to delete cluster ${name} in project ${projectId}: ${String(error)}`);
             }
         })
     );
@@ -88,34 +165,48 @@ async function main(): Promise<void> {
 
     const allErrors: string[] = [];
 
-    for (const project of testProjects) {
-        console.log(`Cleaning up project: ${project.name} (${project.id})`);
-        if (!project.id) {
-            console.warn(`Skipping project with missing ID: ${project.name}`);
-            continue;
-        }
+    const projectsWithIds = testProjects.filter((p): p is Group & { id: string } => !!p.id);
 
-        // Try to delete all clusters first
-        const clusterErrors = await deleteAllClustersOnStaleProject(apiClient, project.id);
-        allErrors.push(...clusterErrors);
+    // Phase 1: Delete all workspaces and clusters in parallel across all projects
+    await Promise.allSettled(
+        projectsWithIds.map(async (project) => {
+            console.log(`Cleaning up project: ${project.name} (${project.id})`);
+            const workspaceErrors = await deleteAllWorkspacesOnStaleProject(apiClient, project.id);
+            allErrors.push(...workspaceErrors);
+            const clusterErrors = await deleteAllClustersOnStaleProject(apiClient, project.id);
+            allErrors.push(...clusterErrors);
+        })
+    );
 
-        // Try to delete the project
-        try {
-            await apiClient.deleteGroup({
-                params: {
-                    path: {
-                        groupId: project.id,
-                    },
-                },
-            });
-            console.log(`Deleted project: ${project.name} (${project.id})`);
-        } catch (error) {
-            const errorStr = String(error);
-            const errorMessage = `Failed to delete project ${project.name} (${project.id}): ${errorStr}`;
-            console.error(errorMessage);
-            allErrors.push(errorMessage);
-        }
-    }
+    // Phase 2: Wait for clusters to terminate, then delete projects in parallel
+    await Promise.allSettled(
+        projectsWithIds.map(async (project) => {
+            // Wait for clusters to be fully deleted (up to 300s)
+            for (let i = 0; i < 300; i++) {
+                try {
+                    const remaining = await apiClient
+                        .listClusters({ params: { path: { groupId: project.id } } })
+                        .then((res) => res.results || []);
+                    if (remaining.length === 0) {
+                        break;
+                    }
+                    await sleep(1000);
+                } catch {
+                    break;
+                }
+            }
+            try {
+                await apiClient.deleteGroup({
+                    params: { path: { groupId: project.id } },
+                });
+                console.log(`Deleted project: ${project.name} (${project.id})`);
+            } catch (error) {
+                const errorMessage = `Failed to delete project ${project.name} (${project.id}): ${String(error)}`;
+                console.error(errorMessage);
+                allErrors.push(errorMessage);
+            }
+        })
+    );
 
     if (allErrors.length > 0) {
         const errorList = allErrors.map((err, i) => `${i + 1}. ${err}`).join("\n");

src/common/config/userConfig.ts

@@ -78,6 +78,8 @@ const ServerConfigSchema = z4.object({
             "drop-collection",
             "delete-many",
             "drop-index",
+            "atlas-streams-manage",
+            "atlas-streams-teardown",
         ])
         .describe(
             "An array of tool names that require user confirmation before execution. Requires the client to support elicitation."

src/common/schemas.ts

@@ -1,4 +1,4 @@
-export const previewFeatureValues = ["search", "mcpUI"] as const;
+export const previewFeatureValues = ["search", "mcpUI", "streams"] as const;
 export type PreviewFeature = (typeof previewFeatureValues)[number];
 
 export const monitoringServerFeatureValues = ["health-check", "metrics"] as const;

src/elicitation.ts

@@ -1,6 +1,12 @@
 import type { ElicitRequestFormParams } from "@modelcontextprotocol/sdk/types.js";
 import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
 
+export type ElicitedInputResult =
+    | { accepted: true; fields: Record<string, string> }
+    | { accepted: false; fields?: undefined };
+
+const ELICITATION_TIMEOUT_MS = 300_000; // 5 minutes for user interaction
+
 export class Elicitation {
     private readonly server: McpServer["server"];
     constructor({ server }: { server: McpServer["server"] }) {
@@ -26,14 +32,56 @@ export class Elicitation {
             return true;
         }
 
-        const result = await this.server.elicitInput({
-            mode: "form",
-            message,
-            requestedSchema: Elicitation.CONFIRMATION_SCHEMA,
-        });
+        const result = await this.server.elicitInput(
+            {
+                mode: "form",
+                message,
+                requestedSchema: Elicitation.CONFIRMATION_SCHEMA,
+            },
+            { timeout: ELICITATION_TIMEOUT_MS }
+        );
         return result.action === "accept" && result.content?.confirmation === "Yes";
     }
 
+    /**
+     * Requests structured input from the user via a form.
+     * Returns the accepted fields, or { accepted: false } if the client doesn't
+     * support elicitation or the user declined.
+     *
+     * @param message - The message/title to display in the form.
+     * @param schema - A JSON Schema describing the fields to collect.
+     * @returns The user-provided values keyed by field name, or null if declined/unsupported.
+     */
+    public async requestInput(
+        message: string,
+        schema: ElicitRequestFormParams["requestedSchema"]
+    ): Promise<ElicitedInputResult> {
+        if (!this.supportsElicitation()) {
+            return { accepted: false };
+        }
+
+        const result = await this.server.elicitInput(
+            {
+                mode: "form",
+                message,
+                requestedSchema: schema,
+            },
+            { timeout: ELICITATION_TIMEOUT_MS }
+        );
+
+        if (result.action !== "accept" || !result.content) {
+            return { accepted: false };
+        }
+
+        const fields: Record<string, string> = {};
+        for (const [key, value] of Object.entries(result.content)) {
+            if (typeof value === "string") {
+                fields[key] = value;
+            }
+        }
+        return { accepted: true, fields };
+    }
+
     /**
      * The schema for the confirmation question.
      * TODO: In the future would be good to use Zod 4's toJSONSchema() to generate the schema.

src/telemetry/types.ts

@@ -147,6 +147,7 @@ export type TelemetryToolMetadata =
     | AtlasMetadata
     | ConnectionMetadata
     | PerfAdvisorToolMetadata
+    | StreamsToolMetadata
     | AutoEmbeddingsUsageMetadata;
 
 export type AtlasMetadata = {
@@ -169,6 +170,11 @@ export type PerfAdvisorToolMetadata = AtlasMetadata &
         operations: string[];
     };
 
+export type StreamsToolMetadata = AtlasMetadata & {
+    action?: string;
+    resource?: string;
+};
+
 export type AutoEmbeddingsUsageMetadata = ConnectionMetadata & {
     /**
      * Indicates which component generated the embeddings.