feat(cli): add MongoDB aggregate query surface (CLI + MCP)

docs-site/content/docs/cli-reference/ktx-mongo-query.mdx

@@ -0,0 +1,103 @@
+---
+title: "ktx mongo-query"
+description: "Fetch rows from a MongoDB connection by running an aggregation pipeline."
+---
+
+Run a MongoDB aggregation pipeline against a `mongodb` connection in your
+**ktx** project. MongoDB is not a SQL source, so `ktx sql` refuses a mongodb
+connection — `ktx mongo-query` is the row-fetch path for MongoDB, and the
+`mongo_query` MCP tool is its agent-facing equivalent.
+
+## Command signature
+
+Use `ktx mongo-query` with a required connection id, a required collection
+name, and a positional aggregation pipeline given as a JSON array of stage
+objects.
+
+```bash
+ktx mongo-query '<pipeline-json>' --connection <id> --collection <name> [options]
+```
+
+## Options
+
+| Flag | Description | Default |
+|------|-------------|---------|
+| `-c`, `--connection <id>` | **ktx** database connection id. Required. | - |
+| `--collection <name>` | Collection to query. Required. | - |
+| `--database <name>` | Database name. | Connection's first configured database |
+| `--limit <n>` | Maximum documents to return. Must be between `1` and `10000`. | `1000` |
+| `--output <mode>` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` on a terminal, `plain` when piped or in CI (override with `KTX_OUTPUT`) |
+| `--json` | Shortcut for `--output=json` (overrides `--output`). | `false` |
+
+## Examples
+
+Quote the pipeline JSON in shell scripts and when it contains spaces or
+punctuation.
+
+```bash
+# Match and project fields from the business collection
+ktx mongo-query '[{"$match":{"city":"Indianapolis"}},{"$project":{"business_id":1,"city":1}}]' \
+  --connection mongo --collection business --limit 500
+
+# Query a non-default database
+ktx mongo-query '[{"$match":{"status":"open"}}]' \
+  --connection mongo \
+  --collection orders \
+  --database analytics
+
+# Print JSON for agents or scripts
+ktx mongo-query '[{"$match":{"status":"open"}}]' \
+  --connection mongo \
+  --collection orders \
+  --json
+
+# Print TSV rows
+ktx mongo-query '[{"$limit":10}]' \
+  -c mongo \
+  --collection orders \
+  --output plain
+```
+
+## Output
+
+Pretty output prints aligned columns and a final row count.
+
+```text
+business_id  city
+-----------  ------------
+1001         Indianapolis
+1002         Indianapolis
+
+2 rows
+```
+
+Plain output prints a TSV header row followed by TSV data rows.
+
+```text
+business_id	city
+1001	Indianapolis
+1002	Indianapolis
+```
+
+JSON output preserves connection id, headers, rows, and row count.
+
+```json
+{
+  "connectionId": "mongo",
+  "headers": ["business_id", "city"],
+  "rows": [
+    [1001, "Indianapolis"],
+    [1002, "Indianapolis"]
+  ],
+  "rowCount": 2
+}
+```
+
+## Common errors
+
+| Error | Cause | Recovery |
+|-------|-------|----------|
+| `must be a JSON aggregation pipeline array` | The pipeline argument is not valid JSON. | Pass a JSON array of pipeline-stage objects. |
+| `must be a JSON array of pipeline-stage objects` | The parsed JSON is not an array of objects. | Wrap each stage as an object inside a JSON array, e.g. `[{"$match":{...}}]`. |
+| `Connection "<id>" is not configured in ktx.yaml` | The connection id is wrong or missing from the project. | Run `ktx connection list` and retry with an exact id. |
+| `driver "<driver>" is not a MongoDB connection` | The connection id resolves to a non-mongodb driver. | Use a `mongodb` connection id, or use `ktx sql` for SQL-queryable drivers. |

docs-site/content/docs/cli-reference/ktx-sql.mdx

@@ -24,7 +24,7 @@ JSON.
 |------|-------------|---------|
 | `-c`, `--connection <id>` | **ktx** database connection id. Required. | - |
 | `--max-rows <n>` | Maximum rows to return. Must be between `1` and `10000`. | `1000` |
-| `--output <mode>` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` |
+| `--output <mode>` | Output mode: `pretty`, `plain` (TSV), or `json`. | `pretty` on a terminal, `plain` when piped or in CI (override with `KTX_OUTPUT`) |
 | `--json` | Shortcut for `--output=json` (overrides `--output`). | `false` |
 
 ## Examples

docs-site/content/docs/cli-reference/meta.json

@@ -8,6 +8,7 @@
     "ktx-ingest",
     "ktx-sl",
     "ktx-sql",
+    "ktx-mongo-query",
     "ktx-wiki",
     "ktx-status",
     "ktx-mcp",

docs-site/content/docs/integrations/primary-sources.mdx

@@ -735,6 +735,7 @@ nullability from how often the field is present:
 | Table sampling | Yes | Reads the most recent documents |
 | Nested analysis | Yes | Sub-documents and arrays modeled as opaque `json` |
 | Read-only SQL (`ktx sql`) | No | MongoDB is not a SQL source |
+| Row queries (`ktx mongo-query`) | Yes | Runs a read-only aggregation pipeline against a collection |
 
 ### Dialect notes
 
@@ -746,6 +747,13 @@ nullability from how often the field is present:
 - `sample_size` trades inference coverage for speed; raise it for collections
   with highly variable documents
 
+### Querying rows
+
+MongoDB is not SQL-queryable, so `ktx sql` refuses a mongodb connection. Use
+[`ktx mongo-query`](/docs/cli-reference/ktx-mongo-query) to fetch rows by
+running a MongoDB aggregation pipeline, or the `mongo_query` MCP tool for the
+same capability from an agent client.
+
 ## Common errors
 
 | Error or symptom | Likely cause | Recovery |

packages/cli/src/cli-program.ts

@@ -8,6 +8,7 @@ import { registerConnectionCommands } from './commands/connection-commands.js';
 import { registerIngestCommands } from './commands/ingest-commands.js';
 import { registerWikiCommands } from './commands/knowledge-commands.js';
 import { registerMcpCommands } from './commands/mcp-commands.js';
+import { registerMongoQueryCommands } from './commands/mongo-query-commands.js';
 import { registerSetupCommands } from './commands/setup-commands.js';
 import { registerSlCommands } from './commands/sl-commands.js';
 import { registerSqlCommands } from './commands/sql-commands.js';
@@ -505,6 +506,7 @@ export function buildKtxProgram(options: BuildKtxProgramOptions): Command {
   registerWikiCommands(program, context);
   registerSlCommands(program, context);
   registerSqlCommands(program, context);
+  registerMongoQueryCommands(program, context);
   registerStatusCommands(program, context);
   registerMcpCommands(program, context);
   registerAdminCommands(program, context);

packages/cli/src/cli-runtime.ts

@@ -4,6 +4,7 @@ import type { KtxConnectionArgs } from './connection.js';
 import type { KtxAdminReindexArgs } from './admin-reindex.js';
 import type { KtxDoctorArgs } from './doctor.js';
 import type { KtxKnowledgeArgs } from './knowledge.js';
+import type { KtxMongoQueryArgs } from './mongo-query.js';
 import type { KtxPublicIngestArgs } from './public-ingest.js';
 import type { KtxRuntimeArgs } from './runtime.js';
 import type { KtxSetupArgs } from './setup.js';
@@ -39,6 +40,7 @@ export interface KtxCliDeps {
   knowledge?: (args: KtxKnowledgeArgs, io: KtxCliIo) => Promise<number>;
   sl?: (args: KtxSlArgs, io: KtxCliIo) => Promise<number>;
   sql?: (args: KtxSqlArgs, io: KtxCliIo) => Promise<number>;
+  mongoQuery?: (args: KtxMongoQueryArgs, io: KtxCliIo) => Promise<number>;
   mcp?: {
     startDaemon?: typeof import('./managed-mcp-daemon.js').startKtxMcpDaemon;
     stopDaemon?: typeof import('./managed-mcp-daemon.js').stopKtxMcpDaemon;

packages/cli/src/commands/mongo-query-commands.ts

@@ -0,0 +1,85 @@
+import { type Command, InvalidArgumentError, Option } from '@commander-js/extra-typings';
+import { type KtxCliCommandContext, resolveCommandProjectDir } from '../cli-program.js';
+import { profileMark } from '../startup-profile.js';
+
+profileMark('module:commands/mongo-query-commands');
+
+const DEFAULT_LIMIT = 1000;
+const LIMIT_CAP = 10_000;
+
+/** @internal exported only for unit testing */
+export function parseLimitOption(value: string): number {
+  const parsed = Number(value);
+  if (!Number.isInteger(parsed) || parsed < 1 || parsed > LIMIT_CAP) {
+    throw new InvalidArgumentError(`must be an integer between 1 and ${LIMIT_CAP}`);
+  }
+  return parsed;
+}
+
+/** @internal exported only for unit testing */
+export function parsePipelineArgument(raw: string): Record<string, unknown>[] {
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(raw);
+  } catch {
+    throw new InvalidArgumentError('must be a JSON aggregation pipeline array, e.g. \'[{"$match":{"city":"NY"}}]\'');
+  }
+  if (
+    !Array.isArray(parsed) ||
+    !parsed.every((stage) => typeof stage === 'object' && stage !== null && !Array.isArray(stage))
+  ) {
+    throw new InvalidArgumentError('must be a JSON array of pipeline-stage objects');
+  }
+  return parsed as Record<string, unknown>[];
+}
+
+export function registerMongoQueryCommands(program: Command, context: KtxCliCommandContext): void {
+  program
+    .command('mongo-query')
+    .description('Fetch rows from a MongoDB connection by running an aggregation pipeline')
+    .argument('<pipeline>', 'MongoDB aggregation pipeline as a JSON array, e.g. \'[{"$match":{"city":"NY"}}]\'', parsePipelineArgument)
+    .requiredOption('-c, --connection <id>', 'ktx connection id')
+    .requiredOption('--collection <name>', 'Collection to query')
+    .option('--database <name>', "Database name (defaults to the connection's first configured database)")
+    .option('--limit <n>', 'Maximum documents to return', parseLimitOption, DEFAULT_LIMIT)
+    .addOption(
+      new Option('--output <mode>', 'Output mode: pretty (default), plain (TSV), or json').choices([
+        'pretty',
+        'plain',
+        'json',
+      ]),
+    )
+    .option('--json', 'Shortcut for --output=json (overrides --output)', false)
+    .action(
+      async (
+        pipeline: Record<string, unknown>[],
+        options: {
+          connection: string;
+          collection: string;
+          database?: string;
+          limit: number;
+          output?: 'pretty' | 'plain' | 'json';
+          json?: boolean;
+        },
+        command,
+      ) => {
+        const runner = context.deps.mongoQuery ?? (await import('../mongo-query.js')).runKtxMongoQuery;
+        context.setExitCode(
+          await runner(
+            {
+              projectDir: resolveCommandProjectDir(command),
+              connectionId: options.connection,
+              collection: options.collection,
+              database: options.database,
+              pipeline,
+              limit: options.limit,
+              output: options.output,
+              json: options.json === true,
+              cliVersion: context.packageInfo.version,
+            },
+            context.io,
+          ),
+        );
+      },
+    );
+}

packages/cli/src/connectors/mongodb/connector.ts

@@ -6,6 +6,8 @@ import {
   type KtxColumnSampleInput,
   type KtxColumnSampleResult,
   type KtxConnectorTestResult,
+  type KtxMongoQueryInput,
+  type KtxMongoQueryResult,
   type KtxScanConnector,
   type KtxScanContext,
   type KtxScanInput,
@@ -50,6 +52,12 @@ export interface KtxMongoClient {
   listCollections(databaseName: string): Promise<KtxMongoListedCollection[]>;
   estimatedDocumentCount(databaseName: string, collectionName: string): Promise<number>;
   find(databaseName: string, collectionName: string, options: KtxMongoFindOptions): Promise<KtxMongoDocument[]>;
+  aggregate(
+    databaseName: string,
+    collectionName: string,
+    pipeline: readonly Record<string, unknown>[],
+    options: { limit: number },
+  ): Promise<KtxMongoDocument[]>;
   ping(databaseName: string): Promise<void>;
   close(): Promise<void>;
 }
@@ -106,6 +114,20 @@ class DefaultMongoClient implements KtxMongoClient {
       .toArray() as Promise<KtxMongoDocument[]>;
   }
 
+  async aggregate(
+    databaseName: string,
+    collectionName: string,
+    pipeline: readonly Record<string, unknown>[],
+    options: { limit: number },
+  ): Promise<KtxMongoDocument[]> {
+    const client = await this.connectedClient();
+    return client
+      .db(databaseName)
+      .collection(collectionName)
+      .aggregate([...pipeline, { $limit: options.limit }], { maxTimeMS: SAMPLE_MAX_TIME_MS })
+      .toArray() as Promise<KtxMongoDocument[]>;
+  }
+
   async ping(databaseName: string): Promise<void> {
     const client = await this.connectedClient();
     await client.db(databaseName).command({ ping: 1 });
@@ -188,6 +210,15 @@ function unionDocumentKeys(documents: readonly KtxMongoDocument[]): string[] {
   return keys;
 }
 
+// MongoDB aggregation write stages ($out/$merge) persist results; mongo_query is read-only.
+function assertReadOnlyPipeline(pipeline: readonly Record<string, unknown>[]): void {
+  for (const stage of pipeline) {
+    if ('$out' in stage || '$merge' in stage) {
+      throw new Error('mongo_query pipelines must be read-only; $out and $merge stages are not allowed.');
+    }
+  }
+}
+
 export class KtxMongoDbScanConnector implements KtxScanConnector {
   readonly id: string;
   readonly driver = 'mongodb' as const;
@@ -362,6 +393,18 @@ export class KtxMongoDbScanConnector implements KtxScanConnector {
     return { values, nullCount, distinctCount: null };
   }
 
+  async executeQuery(input: KtxMongoQueryInput, _ctx: KtxScanContext): Promise<KtxMongoQueryResult> {
+    this.assertConnection(input.connectionId);
+    assertReadOnlyPipeline(input.pipeline);
+    const database = input.database ?? this.databases[0]!;
+    const documents = await this.clientForQuery().aggregate(database, input.collection, input.pipeline, {
+      limit: input.limit,
+    });
+    const headers = unionDocumentKeys(documents);
+    const rows = documents.map((document) => headers.map((header) => normalizeSampleValue(document[header])));
+    return { headers, rows, rowCount: rows.length };
+  }
+
   async listSchemas(): Promise<string[]> {
     return [...this.databases];
   }