diff --git a/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx b/docs-site/content/docs/cli-reference/ktx-mongo-query.mdx
new file mode 100644
index 000000000..a8aaf48bd
--- /dev/null
+++ b/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. |
diff --git a/docs-site/content/docs/cli-reference/ktx-sql.mdx b/docs-site/content/docs/cli-reference/ktx-sql.mdx
index d78864e2a..aa0ef959e 100644
--- a/docs-site/content/docs/cli-reference/ktx-sql.mdx
+++ b/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
diff --git a/docs-site/content/docs/cli-reference/meta.json b/docs-site/content/docs/cli-reference/meta.json
index 2902f2c67..5165c539d 100644
--- a/docs-site/content/docs/cli-reference/meta.json
+++ b/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",
diff --git a/docs-site/content/docs/integrations/primary-sources.mdx b/docs-site/content/docs/integrations/primary-sources.mdx
index b553a3a7e..4617a86c7 100644
--- a/docs-site/content/docs/integrations/primary-sources.mdx
+++ b/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 |
diff --git a/packages/cli/src/cli-program.ts b/packages/cli/src/cli-program.ts
index c30711c85..6fc7affdb 100644
--- a/packages/cli/src/cli-program.ts
+++ b/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);
diff --git a/packages/cli/src/cli-runtime.ts b/packages/cli/src/cli-runtime.ts
index 89c7c11d2..fc59293b2 100644
--- a/packages/cli/src/cli-runtime.ts
+++ b/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;
diff --git a/packages/cli/src/commands/mongo-query-commands.ts b/packages/cli/src/commands/mongo-query-commands.ts
new file mode 100644
index 000000000..206c5df9a
--- /dev/null
+++ b/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,
+          ),
+        );
+      },
+    );
+}
diff --git a/packages/cli/src/connectors/mongodb/connector.ts b/packages/cli/src/connectors/mongodb/connector.ts
index f46d6038e..6d45ed697 100644
--- a/packages/cli/src/connectors/mongodb/connector.ts
+++ b/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];
   }
diff --git a/packages/cli/src/context/mcp/context-tools.ts b/packages/cli/src/context/mcp/context-tools.ts
index 106167e11..e9b0898b8 100644
--- a/packages/cli/src/context/mcp/context-tools.ts
+++ b/packages/cli/src/context/mcp/context-tools.ts
@@ -52,6 +52,7 @@ const toolAnnotations = {
   sl_read_source: { title: 'Semantic Layer Read Source', readOnlyHint: true, idempotentHint: true, openWorldHint: false },
   sl_query: { title: 'Semantic Layer Query', readOnlyHint: true, openWorldHint: false },
   sql_execution: { title: 'SQL Execution', readOnlyHint: true, openWorldHint: false },
+  mongo_query: { title: 'MongoDB Query', readOnlyHint: true, openWorldHint: false },
   sql_dialect_notes: { title: 'SQL Dialect Notes', readOnlyHint: true, idempotentHint: true, openWorldHint: false },
   memory_ingest: { title: 'Memory Ingest', destructiveHint: true, openWorldHint: false },
   memory_ingest_status: { title: 'Memory Ingest Status', readOnlyHint: true, openWorldHint: false },
@@ -75,6 +76,8 @@ const toolDescriptions = {
     'Execute a semantic-layer query and return headers, rows, and total row count, plus correctness notes (e.g. compile-only or fan-out) when relevant. The generated SQL and full query plan are omitted by default; request them with include: ["sql"] and/or include: ["plan"]. Example: sl_query({ connectionId: "warehouse", measures: ["orders.order_count"], dimensions: [{ field: "orders.created_at", granularity: "month" }], include: ["sql"] }).',
   sql_execution:
     'Execute one parser-validated read-only SQL query against a configured ktx connection. Example: sql_execution({ connectionId: "warehouse", sql: "select count(*) from public.orders", maxRows: 100 }).',
+  mongo_query:
+    'Fetch real rows from a live MongoDB connection by running an aggregation pipeline against one collection. MongoDB is not SQL-queryable, so use this — not sql_execution — to filter, project, group, or count Mongo documents. Example: mongo_query({ connectionId: "mongo", collection: "business", pipeline: [{ "$match": { "city": "Indianapolis" } }, { "$project": { "business_id": 1, "city": 1 } }], limit: 500 }).',
   sql_dialect_notes:
     'Return the SQL syntax conventions for the dialect of a ktx connection: fully-qualified table-name form, identifier quoting and case-folding, date/time functions, top-N / window-filtering idiom, and JSON access. Call this before writing raw sql_execution SQL against a connection so the SQL matches that engine. Example: sql_dialect_notes({ connectionId: "warehouse" }).',
   memory_ingest:
@@ -217,6 +220,27 @@ const sqlDialectNotesSchema = z.object({
   connectionId: connectionIdSchema.describe('Connection id whose engine dialect conventions to return.'),
 });
 
+const mongoQuerySchema = z.object({
+  connectionId: connectionIdSchema.describe('MongoDB connection id to query.'),
+  collection: z.string().min(1).describe('Collection to query, e.g. "business".'),
+  database: z
+    .string()
+    .min(1)
+    .optional()
+    .describe('Database name when the connection spans several; defaults to the connection\'s first configured database.'),
+  pipeline: z
+    .array(z.record(z.string(), z.unknown()))
+    .default([])
+    .describe('MongoDB aggregation pipeline stages, e.g. [{ "$match": { "city": "Indianapolis" } }].'),
+  limit: z.number().int().min(1).max(10_000).default(1000).describe('Maximum documents to return.'),
+});
+
+const mongoQueryOutputSchema = z.object({
+  headers: z.array(z.string()),
+  rows: z.array(z.array(z.unknown())),
+  rowCount: z.number(),
+});
+
 const memoryIngestSchema = z.object({
   content: z
     .string()
@@ -971,6 +995,38 @@ export function registerKtxContextTools(deps: RegisterKtxContextToolsDeps): void
     );
   }
 
+  if (ports.mongoQuery) {
+    const mongoQuery = ports.mongoQuery;
+    registerParsedTool(
+      server,
+      'mongo_query',
+      {
+        title: toolAnnotations.mongo_query.title!,
+        description: toolDescriptions.mongo_query,
+        inputSchema: mongoQuerySchema.shape,
+        outputSchema: mongoQueryOutputSchema,
+        annotations: toolAnnotations.mongo_query,
+      },
+      mongoQuerySchema,
+      async (input, context) => {
+        const onProgress = mcpProgressCallback(context);
+        return jsonToolResult(
+          await mongoQuery.execute(
+            {
+              connectionId: input.connectionId,
+              collection: input.collection,
+              ...(input.database ? { database: input.database } : {}),
+              pipeline: input.pipeline,
+              limit: input.limit,
+            },
+            onProgress ? { onProgress } : undefined,
+          ),
+        );
+      },
+      toolTelemetry,
+    );
+  }
+
   if (ports.dialectNotes) {
     const dialectNotes = ports.dialectNotes;
     registerParsedTool(
diff --git a/packages/cli/src/context/mcp/local-project-ports.ts b/packages/cli/src/context/mcp/local-project-ports.ts
index 684c002ca..d1cc28e47 100644
--- a/packages/cli/src/context/mcp/local-project-ports.ts
+++ b/packages/cli/src/context/mcp/local-project-ports.ts
@@ -24,6 +24,7 @@ import { readLocalSlSource } from '../../context/sl/local-sl.js';
 import { assertSafeConnectionId } from '../../context/sl/source-files.js';
 import { assertConfiguredConnectionId } from '../../context/connections/configured-connections.js';
 import { readLocalKnowledgePage, searchLocalKnowledgePages } from '../wiki/local-knowledge.js';
+import { runMongoQuery } from './mongo-query-runner.js';
 import type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxSqlExecutionResponse } from './types.js';
 
 interface CreateLocalProjectMcpContextPortsOptions {
@@ -54,6 +55,12 @@ function throwClassifiedQueryError(error: unknown): never {
   throw new KtxQueryError(error instanceof Error ? error.message : String(error), { cause: error });
 }
 
+function projectHasMongoConnection(project: KtxLocalProject): boolean {
+  return Object.values(project.config.connections).some(
+    (connection) => normalizeConnectionDriver(connection) === 'mongodb',
+  );
+}
+
 async function executeValidatedReadOnlySql(
   project: KtxLocalProject,
   options: CreateLocalProjectMcpContextPortsOptions,
@@ -239,5 +246,18 @@ export function createLocalProjectMcpContextPorts(
     };
   }
 
+  const mongoCreateConnector = options.localScan?.createConnector;
+  if (mongoCreateConnector && projectHasMongoConnection(project)) {
+    ports.mongoQuery = {
+      async execute(input) {
+        try {
+          return await runMongoQuery(mongoCreateConnector, input);
+        } catch (error) {
+          throwClassifiedQueryError(error);
+        }
+      },
+    };
+  }
+
   return ports;
 }
diff --git a/packages/cli/src/context/mcp/mongo-query-runner.ts b/packages/cli/src/context/mcp/mongo-query-runner.ts
new file mode 100644
index 000000000..2d45dc543
--- /dev/null
+++ b/packages/cli/src/context/mcp/mongo-query-runner.ts
@@ -0,0 +1,23 @@
+import { KtxExpectedError } from '../../errors.js';
+import { assertSafeConnectionId } from '../sl/source-files.js';
+import type { KtxMongoQueryInput, KtxMongoQueryResult, KtxScanConnector } from '../scan/types.js';
+
+/** Single Mongo-read execution seam: resolve the connector, guard the driver, run the pipeline. */
+export async function runMongoQuery(
+  createConnector: (connectionId: string) => Promise<KtxScanConnector> | KtxScanConnector,
+  input: KtxMongoQueryInput,
+): Promise<KtxMongoQueryResult> {
+  const connectionId = assertSafeConnectionId(input.connectionId);
+  let connector: KtxScanConnector | null = null;
+  try {
+    connector = await createConnector(connectionId);
+    if (connector.driver !== 'mongodb' || typeof connector.executeQuery !== 'function') {
+      throw new KtxExpectedError(
+        `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`,
+      );
+    }
+    return await connector.executeQuery(input, { runId: 'mongo-query' });
+  } finally {
+    await connector?.cleanup?.();
+  }
+}
diff --git a/packages/cli/src/context/mcp/types.ts b/packages/cli/src/context/mcp/types.ts
index c8dbd480b..f05c49b47 100644
--- a/packages/cli/src/context/mcp/types.ts
+++ b/packages/cli/src/context/mcp/types.ts
@@ -5,6 +5,7 @@ import type { KtxEntityDetailsInput, KtxEntityDetailsResponse } from '../scan/en
 import type { KtxDiscoverDataInput, KtxDiscoverDataResponse } from '../../context/search/discover.js';
 import type { KtxDictionarySearchInput, KtxDictionarySearchResponse } from '../../context/sl/dictionary-search.js';
 import type { SemanticLayerQueryInput } from '../../context/sl/types.js';
+import type { KtxMongoQueryInput, KtxMongoQueryResult } from '../scan/types.js';
 import type { WikiSearchLaneSummary, WikiSearchMatchReason } from '../../context/wiki/types.js';
 
 interface KtxMcpTextContent {
@@ -180,6 +181,11 @@ export interface KtxSqlExecutionMcpPort {
   ): Promise<KtxSqlExecutionResponse>;
 }
 
+/** @internal */
+export interface KtxMongoQueryMcpPort {
+  execute(input: KtxMongoQueryInput, options?: { onProgress?: KtxMcpProgressCallback }): Promise<KtxMongoQueryResult>;
+}
+
 /** @internal */
 export interface KtxDialectNotesMcpPort {
   read(input: { connectionId: string }): Promise<{ connectionId: string; dialect: string; notes: string }>;
@@ -193,6 +199,7 @@ export interface KtxMcpContextPorts {
   dictionarySearch?: KtxDictionarySearchMcpPort;
   discover?: KtxDiscoverDataMcpPort;
   sqlExecution?: KtxSqlExecutionMcpPort;
+  mongoQuery?: KtxMongoQueryMcpPort;
   dialectNotes?: KtxDialectNotesMcpPort;
   memoryIngest?: MemoryIngestPort;
 }
diff --git a/packages/cli/src/context/scan/types.ts b/packages/cli/src/context/scan/types.ts
index bf72558c4..3d39307b5 100644
--- a/packages/cli/src/context/scan/types.ts
+++ b/packages/cli/src/context/scan/types.ts
@@ -293,6 +293,20 @@ export interface KtxReadOnlyQueryInput {
   maxRows?: number;
 }
 
+export interface KtxMongoQueryInput {
+  connectionId: string;
+  collection: string;
+  database?: string;
+  pipeline: readonly Record<string, unknown>[];
+  limit: number;
+}
+
+export interface KtxMongoQueryResult {
+  headers: string[];
+  rows: unknown[][];
+  rowCount: number;
+}
+
 export interface KtxQueryResult {
   headers: string[];
   headerTypes?: string[];
@@ -346,6 +360,7 @@ export interface KtxScanConnector {
   sampleTable?(input: KtxTableSampleInput, ctx: KtxScanContext): Promise<KtxTableSampleResult>;
   columnStats?(input: KtxColumnStatsInput, ctx: KtxScanContext): Promise<KtxColumnStatsResult | null>;
   executeReadOnly?(input: KtxReadOnlyQueryInput, ctx: KtxScanContext): Promise<KtxQueryResult>;
+  executeQuery?(input: KtxMongoQueryInput, ctx: KtxScanContext): Promise<KtxMongoQueryResult>;
   cleanup?(): Promise<void>;
 }
 
diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts
index 236bdf699..740d8763d 100644
--- a/packages/cli/src/index.ts
+++ b/packages/cli/src/index.ts
@@ -34,6 +34,7 @@ export type {
 export { runKtxSetupSourcesStep } from './setup-sources.js';
 export { runKtxRuntime, type KtxRuntimeArgs, type KtxRuntimeDeps } from './runtime.js';
 export { runKtxSql, type KtxSqlArgs, type KtxSqlDeps } from './sql.js';
+export { runKtxMongoQuery, type KtxMongoQueryArgs, type KtxMongoQueryCliDeps } from './mongo-query.js';
 export {
   allocateDaemonPort,
   readManagedPythonDaemonStatus,
diff --git a/packages/cli/src/io/result-table.ts b/packages/cli/src/io/result-table.ts
new file mode 100644
index 000000000..98f301d08
--- /dev/null
+++ b/packages/cli/src/io/result-table.ts
@@ -0,0 +1,59 @@
+import type { KtxCliIo } from '../cli-runtime.js';
+import type { KtxOutputMode } from './mode.js';
+
+export interface KtxResultTable {
+  connectionId: string;
+  headers: string[];
+  headerTypes?: string[];
+  rows: unknown[][];
+  rowCount: number;
+}
+
+/** @internal */
+export function formatValue(value: unknown): string {
+  if (value === null || value === undefined) return '';
+  if (typeof value === 'string') return value;
+  if (typeof value === 'number' || typeof value === 'boolean' || typeof value === 'bigint') return String(value);
+  return JSON.stringify(value);
+}
+
+function printJson(output: KtxResultTable, io: KtxCliIo): void {
+  io.stdout.write(`${JSON.stringify(output, null, 2)}\n`);
+}
+
+function printPlain(output: KtxResultTable, io: KtxCliIo): void {
+  io.stdout.write(`${output.headers.join('\t')}\n`);
+  for (const row of output.rows) {
+    io.stdout.write(`${row.map(formatValue).join('\t')}\n`);
+  }
+}
+
+function printPretty(output: KtxResultTable, io: KtxCliIo): void {
+  const rows = output.rows.map((row) => row.map(formatValue));
+  const widths = output.headers.map((header, index) =>
+    Math.max(header.length, ...rows.map((row) => row[index]?.length ?? 0)),
+  );
+  const renderRow = (cells: string[]): string =>
+    cells.map((cell, index) => cell.padEnd(widths[index] ?? cell.length)).join('  ').trimEnd();
+
+  if (output.headers.length > 0) {
+    io.stdout.write(`${renderRow(output.headers)}\n`);
+    io.stdout.write(`${renderRow(widths.map((width) => '-'.repeat(width)))}\n`);
+  }
+  for (const row of rows) {
+    io.stdout.write(`${renderRow(row)}\n`);
+  }
+  io.stdout.write(`\n${output.rowCount} ${output.rowCount === 1 ? 'row' : 'rows'}\n`);
+}
+
+export function printResultTable(output: KtxResultTable, mode: KtxOutputMode, io: KtxCliIo): void {
+  if (mode === 'json') {
+    printJson(output, io);
+    return;
+  }
+  if (mode === 'plain') {
+    printPlain(output, io);
+    return;
+  }
+  printPretty(output, io);
+}
diff --git a/packages/cli/src/mongo-query.ts b/packages/cli/src/mongo-query.ts
new file mode 100644
index 000000000..bc0e6e8f1
--- /dev/null
+++ b/packages/cli/src/mongo-query.ts
@@ -0,0 +1,100 @@
+import { runMongoQuery } from './context/mcp/mongo-query-runner.js';
+import { resolveConfiguredConnection } from './context/connections/resolve-connection.js';
+import { loadKtxProject, type KtxLocalProject } from './context/project/project.js';
+import type { KtxMongoQueryInput } from './context/scan/types.js';
+import type { KtxCliIo } from './cli-runtime.js';
+import { type KtxOutputMode, resolveOutputMode } from './io/mode.js';
+import { type KtxResultTable, printResultTable } from './io/result-table.js';
+import { createKtxCliScanConnector } from './local-scan-connectors.js';
+import { profileMark } from './startup-profile.js';
+import { isDemoConnection } from './telemetry/demo-detect.js';
+import { emitTelemetryEvent, reportException } from './telemetry/index.js';
+import { collectTelemetryRedactionSecrets } from './telemetry/redaction-secrets.js';
+import { scrubErrorClass } from './telemetry/scrubber.js';
+
+profileMark('module:mongo-query');
+
+export type KtxMongoQueryArgs = KtxMongoQueryInput & {
+  projectDir: string;
+  output?: KtxOutputMode;
+  json?: boolean;
+  cliVersion: string;
+};
+
+export interface KtxMongoQueryCliDeps {
+  loadProject?: typeof loadKtxProject;
+  createScanConnector?: typeof createKtxCliScanConnector;
+}
+
+export async function runKtxMongoQuery(
+  args: KtxMongoQueryArgs,
+  io: KtxCliIo = process,
+  deps: KtxMongoQueryCliDeps = {},
+): Promise<number> {
+  const startedAt = performance.now();
+  let driver = 'unknown';
+  let demoConnection = false;
+  let project: KtxLocalProject | undefined;
+  try {
+    project = await (deps.loadProject ?? loadKtxProject)({ projectDir: args.projectDir });
+    const connection = resolveConfiguredConnection(project.config, args.connectionId);
+    driver = String(connection?.driver ?? 'unknown').toLowerCase();
+    demoConnection = isDemoConnection(args.connectionId, connection);
+
+    const createScanConnector = deps.createScanConnector ?? createKtxCliScanConnector;
+    const result = await runMongoQuery((connectionId) => createScanConnector(project!, connectionId), args);
+
+    const output: KtxResultTable = {
+      connectionId: args.connectionId,
+      headers: result.headers,
+      rows: result.rows,
+      rowCount: result.rowCount,
+    };
+    const mode = resolveOutputMode({ explicit: args.output, json: args.json, io });
+    printResultTable(output, mode, io);
+    await emitTelemetryEvent({
+      name: 'mongo_query_completed',
+      projectDir: args.projectDir,
+      io,
+      fields: {
+        driver,
+        isDemoConnection: demoConnection,
+        stageCount: args.pipeline.length,
+        durationMs: Math.max(0, performance.now() - startedAt),
+        outcome: 'ok',
+      },
+    });
+    return 0;
+  } catch (error) {
+    const errorClass = scrubErrorClass(error);
+    await emitTelemetryEvent({
+      name: 'mongo_query_completed',
+      projectDir: args.projectDir,
+      io,
+      fields: {
+        driver,
+        isDemoConnection: demoConnection,
+        stageCount: args.pipeline.length,
+        durationMs: Math.max(0, performance.now() - startedAt),
+        outcome: 'error',
+        ...(errorClass ? { errorClass } : {}),
+      },
+    });
+    await reportException({
+      error,
+      context: { source: 'mongo-query run', handled: true, fatal: false },
+      projectDir: args.projectDir,
+      io,
+      redactionSecrets: await collectTelemetryRedactionSecrets({
+        project,
+        projectDir: args.projectDir,
+        connectionId: args.connectionId,
+        includeLlm: false,
+        includeEmbeddings: false,
+        env: process.env,
+      }),
+    });
+    io.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
+    return 1;
+  }
+}
diff --git a/packages/cli/src/sql.ts b/packages/cli/src/sql.ts
index fbf059757..bd29245a2 100644
--- a/packages/cli/src/sql.ts
+++ b/packages/cli/src/sql.ts
@@ -9,6 +9,7 @@ import { sqlAnalysisDialectForDriver } from './context/sql-analysis/dialect.js';
 import type { SqlAnalysisDialect, SqlAnalysisPort } from './context/sql-analysis/ports.js';
 import type { KtxCliIo } from './cli-runtime.js';
 import { type KtxOutputMode, resolveOutputMode } from './io/mode.js';
+import { type KtxResultTable, printResultTable } from './io/result-table.js';
 import { createKtxCliScanConnector } from './local-scan-connectors.js';
 import { createManagedDaemonSqlAnalysisPort } from './managed-python-http.js';
 import { profileMark } from './startup-profile.js';
@@ -39,14 +40,6 @@ export interface KtxSqlDeps {
   executeFederated?: typeof executeFederatedQuery;
 }
 
-interface SqlExecutionOutput {
-  connectionId: string;
-  headers: string[];
-  headerTypes?: string[];
-  rows: unknown[][];
-  rowCount: number;
-}
-
 function queryVerb(sql: string): 'select' | 'explain' | 'show' | 'with' | 'other' {
   const first = sql.trim().split(/\s+/, 1)[0]?.toLowerCase();
   if (first === 'select' || first === 'explain' || first === 'show' || first === 'with') {
@@ -68,55 +61,7 @@ async function safeReferencedTableCount(
   }
 }
 
-function formatValue(value: unknown): string {
-  if (value === null || value === undefined) return '';
-  if (typeof value === 'string') return value;
-  if (typeof value === 'number' || typeof value === 'boolean' || typeof value === 'bigint') return String(value);
-  return JSON.stringify(value);
-}
-
-function printJson(output: SqlExecutionOutput, io: KtxCliIo): void {
-  io.stdout.write(`${JSON.stringify(output, null, 2)}\n`);
-}
-
-function printPlain(output: SqlExecutionOutput, io: KtxCliIo): void {
-  io.stdout.write(`${output.headers.join('\t')}\n`);
-  for (const row of output.rows) {
-    io.stdout.write(`${row.map(formatValue).join('\t')}\n`);
-  }
-}
-
-function printPretty(output: SqlExecutionOutput, io: KtxCliIo): void {
-  const rows = output.rows.map((row) => row.map(formatValue));
-  const widths = output.headers.map((header, index) =>
-    Math.max(header.length, ...rows.map((row) => row[index]?.length ?? 0)),
-  );
-  const renderRow = (cells: string[]): string =>
-    cells.map((cell, index) => cell.padEnd(widths[index] ?? cell.length)).join('  ').trimEnd();
-
-  if (output.headers.length > 0) {
-    io.stdout.write(`${renderRow(output.headers)}\n`);
-    io.stdout.write(`${renderRow(widths.map((width) => '-'.repeat(width)))}\n`);
-  }
-  for (const row of rows) {
-    io.stdout.write(`${renderRow(row)}\n`);
-  }
-  io.stdout.write(`\n${output.rowCount} ${output.rowCount === 1 ? 'row' : 'rows'}\n`);
-}
-
-function printSqlResult(output: SqlExecutionOutput, mode: KtxSqlOutputMode, io: KtxCliIo): void {
-  if (mode === 'json') {
-    printJson(output, io);
-    return;
-  }
-  if (mode === 'plain') {
-    printPlain(output, io);
-    return;
-  }
-  printPretty(output, io);
-}
-
-function resultOutput(connectionId: string, result: KtxSqlQueryExecutionResult): SqlExecutionOutput {
+function resultOutput(connectionId: string, result: KtxSqlQueryExecutionResult): KtxResultTable {
   return {
     connectionId,
     headers: result.headers,
@@ -168,7 +113,7 @@ export async function runKtxSql(args: KtxSqlArgs, io: KtxCliIo = process, deps:
     });
     const referencedTableCount = await safeReferencedTableCount(analysisPort, args.sql, dialect);
     const mode = resolveOutputMode({ explicit: args.output, json: args.json, io });
-    printSqlResult(resultOutput(args.connectionId, result), mode, io);
+    printResultTable(resultOutput(args.connectionId, result), mode, io);
     await emitTelemetryEvent({
       name: 'sql_completed',
       projectDir: args.projectDir,
diff --git a/packages/cli/src/telemetry/events.schema.json b/packages/cli/src/telemetry/events.schema.json
index 405f02400..0bd22bb3d 100644
--- a/packages/cli/src/telemetry/events.schema.json
+++ b/packages/cli/src/telemetry/events.schema.json
@@ -144,6 +144,18 @@
         "errorClass"
       ]
     },
+    {
+      "name": "mongo_query_completed",
+      "description": "Emitted after ktx mongo-query completes validation and execution.",
+      "fields": [
+        "driver",
+        "isDemoConnection",
+        "stageCount",
+        "durationMs",
+        "outcome",
+        "errorClass"
+      ]
+    },
     {
       "name": "wiki_query_completed",
       "description": "Emitted after a wiki query completes.",
@@ -1056,6 +1068,77 @@
       ],
       "additionalProperties": false
     },
+    "mongo_query_completed": {
+      "$schema": "https://json-schema.org/draft/2020-12/schema",
+      "type": "object",
+      "properties": {
+        "cliVersion": {
+          "type": "string"
+        },
+        "nodeVersion": {
+          "type": "string"
+        },
+        "osPlatform": {
+          "type": "string"
+        },
+        "osRelease": {
+          "type": "string"
+        },
+        "arch": {
+          "type": "string"
+        },
+        "runtime": {
+          "type": "string",
+          "enum": [
+            "node",
+            "daemon-py"
+          ]
+        },
+        "isCi": {
+          "type": "boolean"
+        },
+        "driver": {
+          "type": "string"
+        },
+        "isDemoConnection": {
+          "type": "boolean"
+        },
+        "stageCount": {
+          "type": "integer",
+          "minimum": 0,
+          "maximum": 9007199254740991
+        },
+        "durationMs": {
+          "type": "number",
+          "minimum": 0
+        },
+        "outcome": {
+          "type": "string",
+          "enum": [
+            "ok",
+            "error"
+          ]
+        },
+        "errorClass": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "cliVersion",
+        "nodeVersion",
+        "osPlatform",
+        "osRelease",
+        "arch",
+        "runtime",
+        "isCi",
+        "driver",
+        "isDemoConnection",
+        "stageCount",
+        "durationMs",
+        "outcome"
+      ],
+      "additionalProperties": false
+    },
     "wiki_query_completed": {
       "$schema": "https://json-schema.org/draft/2020-12/schema",
       "type": "object",
diff --git a/packages/cli/src/telemetry/events.ts b/packages/cli/src/telemetry/events.ts
index 6d311ac53..84b8617b8 100644
--- a/packages/cli/src/telemetry/events.ts
+++ b/packages/cli/src/telemetry/events.ts
@@ -146,6 +146,17 @@ const sqlCompletedSchema = telemetryCommonEnvelopeSchema
   })
   .strict();
 
+const mongoQueryCompletedSchema = telemetryCommonEnvelopeSchema
+  .extend({
+    driver: z.string(),
+    isDemoConnection: z.boolean(),
+    stageCount: z.number().int().nonnegative(),
+    durationMs: z.number().nonnegative(),
+    outcome: outcomeSchema,
+    errorClass: z.string().optional(),
+  })
+  .strict();
+
 const wikiQueryCompletedSchema = telemetryCommonEnvelopeSchema
   .extend({
     queryLength: z.number().int().nonnegative(),
@@ -230,6 +241,7 @@ export const telemetryEventSchemas = {
   sl_validate_completed: slValidateCompletedSchema,
   sl_query_completed: slQueryCompletedSchema,
   sql_completed: sqlCompletedSchema,
+  mongo_query_completed: mongoQueryCompletedSchema,
   wiki_query_completed: wikiQueryCompletedSchema,
   mcp_request_completed: mcpRequestCompletedSchema,
   daemon_started: daemonStartedSchema,
@@ -342,6 +354,11 @@ export const telemetryEventCatalog = [
       'errorClass',
     ],
   },
+  {
+    name: 'mongo_query_completed',
+    description: 'Emitted after ktx mongo-query completes validation and execution.',
+    fields: ['driver', 'isDemoConnection', 'stageCount', 'durationMs', 'outcome', 'errorClass'],
+  },
   {
     name: 'wiki_query_completed',
     description: 'Emitted after a wiki query completes.',
diff --git a/packages/cli/test/commands/mongo-query-commands.test.ts b/packages/cli/test/commands/mongo-query-commands.test.ts
new file mode 100644
index 000000000..c04d988d4
--- /dev/null
+++ b/packages/cli/test/commands/mongo-query-commands.test.ts
@@ -0,0 +1,59 @@
+import { InvalidArgumentError } from '@commander-js/extra-typings';
+import { describe, expect, it } from 'vitest';
+import { parseLimitOption, parsePipelineArgument } from '../../src/commands/mongo-query-commands.js';
+
+describe('parsePipelineArgument', () => {
+  it('parses a valid JSON array of pipeline-stage objects', () => {
+    expect(parsePipelineArgument('[{"$match":{"city":"NY"}},{"$limit":10}]')).toEqual([
+      { $match: { city: 'NY' } },
+      { $limit: 10 },
+    ]);
+  });
+
+  it('parses an empty pipeline array', () => {
+    expect(parsePipelineArgument('[]')).toEqual([]);
+  });
+
+  it('throws InvalidArgumentError on invalid JSON', () => {
+    expect(() => parsePipelineArgument('{not json')).toThrow(InvalidArgumentError);
+  });
+
+  it('throws InvalidArgumentError when the value is not an array', () => {
+    expect(() => parsePipelineArgument('{}')).toThrow(InvalidArgumentError);
+  });
+
+  it('throws InvalidArgumentError when the array contains a non-object stage', () => {
+    expect(() => parsePipelineArgument('[{"$match":{}}, "not-a-stage"]')).toThrow(InvalidArgumentError);
+  });
+
+  it('throws InvalidArgumentError when the array contains a nested array stage', () => {
+    expect(() => parsePipelineArgument('[[1,2,3]]')).toThrow(InvalidArgumentError);
+  });
+
+  it('throws InvalidArgumentError when the array contains null', () => {
+    expect(() => parsePipelineArgument('[null]')).toThrow(InvalidArgumentError);
+  });
+});
+
+describe('parseLimitOption', () => {
+  it('accepts the boundary values 1 and 10000', () => {
+    expect(parseLimitOption('1')).toBe(1);
+    expect(parseLimitOption('10000')).toBe(10000);
+  });
+
+  it('rejects a non-numeric value', () => {
+    expect(() => parseLimitOption('abc')).toThrow(InvalidArgumentError);
+  });
+
+  it('rejects a non-integer value', () => {
+    expect(() => parseLimitOption('1.5')).toThrow(InvalidArgumentError);
+  });
+
+  it('rejects a value below 1', () => {
+    expect(() => parseLimitOption('0')).toThrow(InvalidArgumentError);
+  });
+
+  it('rejects a value above the cap', () => {
+    expect(() => parseLimitOption('10001')).toThrow(InvalidArgumentError);
+  });
+});
diff --git a/packages/cli/test/connectors/mongodb/connector.test.ts b/packages/cli/test/connectors/mongodb/connector.test.ts
index 43d60013d..e46d4bc2b 100644
--- a/packages/cli/test/connectors/mongodb/connector.test.ts
+++ b/packages/cli/test/connectors/mongodb/connector.test.ts
@@ -40,6 +40,7 @@ function fakeClientFactory(): { factory: KtxMongoClientFactory; client: KtxMongo
       collectionName === 'users' ? 2 : 1,
     ),
     find: vi.fn(async (_databaseName: string, collectionName: string) => DOCUMENTS[collectionName] ?? []),
+    aggregate: vi.fn(async (_databaseName: string, collectionName: string) => DOCUMENTS[collectionName] ?? []),
     ping: vi.fn(async () => undefined),
     close: vi.fn(async () => undefined),
   };
@@ -184,6 +185,7 @@ describe('KtxMongoDbScanConnector.introspect', () => {
         return 2;
       }),
       find: vi.fn(async (_db: string, name: string) => DOCUMENTS[name] ?? DOCUMENTS.users!),
+      aggregate: vi.fn(async (_db: string, name: string) => DOCUMENTS[name] ?? DOCUMENTS.users!),
       ping: vi.fn(async () => undefined),
       close: vi.fn(async () => undefined),
     };
@@ -239,3 +241,77 @@ describe('ktx sql against a MongoDB connection', () => {
     ).rejects.toThrow(/does not support read-only SQL execution/);
   });
 });
+
+describe('KtxMongoDbScanConnector.executeQuery', () => {
+  it('runs an aggregation pipeline and returns normalized rows keyed by the union of document fields', async () => {
+    const { factory, client } = fakeClientFactory();
+    const result = await connector(baseConnection, factory).executeQuery(
+      {
+        connectionId: 'mongo-prod',
+        collection: 'users',
+        pipeline: [{ $match: { age: { $gte: 18 } } }],
+        limit: 50,
+      },
+      { runId: 't' },
+    );
+
+    expect(client.aggregate).toHaveBeenCalledWith('app', 'users', [{ $match: { age: { $gte: 18 } } }], {
+      limit: 50,
+    });
+    expect(result.headers).toEqual(['_id', 'email', 'age', 'address']);
+    // _id is a BSON wrapper → String(...); address is a sub-document → JSON string.
+    expect(result.rows[0]).toEqual(['a1', 'a@x.com', 31, JSON.stringify({ city: 'NY' })]);
+    // Second document is missing age and address → null-filled to keep rows rectangular.
+    expect(result.rows[1]).toEqual(['a2', 'b@x.com', null, null]);
+    expect(result.rowCount).toBe(2);
+  });
+
+  it('rejects a query whose connection id does not match this connector', async () => {
+    const { factory } = fakeClientFactory();
+    await expect(
+      connector(baseConnection, factory).executeQuery(
+        { connectionId: 'other', collection: 'users', pipeline: [], limit: 10 },
+        { runId: 't' },
+      ),
+    ).rejects.toThrow(/cannot serve connection other/);
+  });
+
+  it('rejects a pipeline containing a $out write stage', async () => {
+    const { factory } = fakeClientFactory();
+    await expect(
+      connector(baseConnection, factory).executeQuery(
+        { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $out: 'exfiltrated' }], limit: 10 },
+        { runId: 't' },
+      ),
+    ).rejects.toThrow(/must be read-only/);
+  });
+
+  it('rejects a pipeline containing a $merge write stage', async () => {
+    const { factory } = fakeClientFactory();
+    await expect(
+      connector(baseConnection, factory).executeQuery(
+        { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $merge: { into: 'x' } }], limit: 10 },
+        { runId: 't' },
+      ),
+    ).rejects.toThrow(/must be read-only/);
+  });
+
+  it('runs against the requested database when input.database is given', async () => {
+    const { factory, client } = fakeClientFactory();
+    await connector(baseConnection, factory).executeQuery(
+      { connectionId: 'mongo-prod', collection: 'orders', database: 'analytics', pipeline: [], limit: 5 },
+      { runId: 't' },
+    );
+    expect(client.aggregate).toHaveBeenCalledWith('analytics', 'orders', [], { limit: 5 });
+  });
+
+  it('returns an empty result with no headers when the pipeline matches no documents', async () => {
+    const { factory, client } = fakeClientFactory();
+    (client.aggregate as ReturnType<typeof vi.fn>).mockResolvedValueOnce([]);
+    const result = await connector(baseConnection, factory).executeQuery(
+      { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $match: { age: { $gte: 999 } } }], limit: 50 },
+      { runId: 't' },
+    );
+    expect(result).toEqual({ headers: [], rows: [], rowCount: 0 });
+  });
+});
diff --git a/packages/cli/test/context/mcp/server.test.ts b/packages/cli/test/context/mcp/server.test.ts
index 6a7e756a4..43f9b3204 100644
--- a/packages/cli/test/context/mcp/server.test.ts
+++ b/packages/cli/test/context/mcp/server.test.ts
@@ -21,11 +21,13 @@ import type {
   KtxKnowledgeMcpPort,
   KtxMcpContextPorts,
   KtxMcpToolHandlerContext,
+  KtxMongoQueryMcpPort,
   KtxSemanticLayerMcpPort,
   KtxSqlExecutionMcpPort,
   KtxSqlExecutionResponse,
   MemoryIngestPort,
 } from '../../../src/context/mcp/types.js';
+import type { KtxMongoQueryResult } from '../../../src/context/scan/types.js';
 
 const reportExceptionMock = vi.hoisted(() => vi.fn(async () => {}));
 
@@ -495,6 +497,56 @@ describe('createKtxMcpServer', () => {
     );
   });
 
+  it('registers mongo_query when the host provides a MongoDB query port', async () => {
+    const fake = makeFakeServer();
+    const response: KtxMongoQueryResult = {
+      headers: ['_id', 'city'],
+      rows: [
+        ['a1', 'Indianapolis'],
+        ['a2', 'Indianapolis'],
+      ],
+      rowCount: 2,
+    };
+    const mongoQuery: KtxMongoQueryMcpPort = {
+      execute: vi.fn<KtxMongoQueryMcpPort['execute']>().mockResolvedValue(response),
+    };
+
+    createKtxMcpServer({
+      server: fake.server,
+      userContext: { userId: 'local-user' },
+      contextTools: {
+        mongoQuery,
+      },
+    });
+
+    expect(fake.tools.map((tool) => tool.name)).toEqual(['mongo_query']);
+    await expect(
+      getTool(fake.tools, 'mongo_query').handler({
+        connectionId: 'mongo',
+        collection: 'business',
+        pipeline: [{ $match: { city: 'Indianapolis' } }],
+        limit: 100,
+      }),
+    ).resolves.toEqual({
+      content: [
+        {
+          type: 'text',
+          text: JSON.stringify(response),
+        },
+      ],
+      structuredContent: response,
+    });
+    expect(mongoQuery.execute).toHaveBeenCalledWith(
+      {
+        connectionId: 'mongo',
+        collection: 'business',
+        pipeline: [{ $match: { city: 'Indianapolis' } }],
+        limit: 100,
+      },
+      undefined,
+    );
+  });
+
   it('registers entity_details when the host provides an entity-details port', async () => {
     const fake = makeFakeServer();
     const entityDetails: KtxEntityDetailsMcpPort = {
diff --git a/packages/cli/test/io/result-table.test.ts b/packages/cli/test/io/result-table.test.ts
new file mode 100644
index 000000000..9129265ef
--- /dev/null
+++ b/packages/cli/test/io/result-table.test.ts
@@ -0,0 +1,70 @@
+import { describe, expect, it } from 'vitest';
+import { formatValue, printResultTable } from '../../src/io/result-table.js';
+import type { KtxCliIo } from '../../src/cli-runtime.js';
+
+function captureIo(): { io: KtxCliIo; out: () => string } {
+  let buffer = '';
+  const io = {
+    stdout: { write: (s: string) => { buffer += s; return true; } },
+    stderr: { write: () => true },
+  } as unknown as KtxCliIo;
+  return { io, out: () => buffer };
+}
+
+describe('formatValue', () => {
+  it('renders null/undefined as empty, scalars as strings, objects as JSON', () => {
+    expect(formatValue(null)).toBe('');
+    expect(formatValue(undefined)).toBe('');
+    expect(formatValue('x')).toBe('x');
+    expect(formatValue(42)).toBe('42');
+    expect(formatValue(true)).toBe('true');
+    expect(formatValue(10n)).toBe('10');
+    expect(formatValue({ a: 1 })).toBe(JSON.stringify({ a: 1 }));
+  });
+});
+
+describe('printResultTable', () => {
+  const table = { connectionId: 'mongo', headers: ['_id', 'city'], rows: [['a1', 'NY']], rowCount: 1 };
+
+  it('json mode prints the structured payload', () => {
+    const { io, out } = captureIo();
+    printResultTable(table, 'json', io);
+    expect(JSON.parse(out())).toEqual(table);
+  });
+
+  it('plain mode prints tab-separated headers and rows', () => {
+    const { io, out } = captureIo();
+    printResultTable(table, 'plain', io);
+    expect(out()).toBe('_id\tcity\na1\tNY\n');
+  });
+
+  it('pretty mode prints a header rule and a singular row count', () => {
+    const { io, out } = captureIo();
+    printResultTable(table, 'pretty', io);
+    expect(out()).toContain('_id');
+    expect(out()).toContain('1 row\n');
+  });
+
+  it('pretty mode aligns multiple rows and pluralizes the row count', () => {
+    const { io, out } = captureIo();
+    const multi = {
+      connectionId: 'mongo',
+      headers: ['id', 'city'],
+      rows: [
+        ['1', 'NY'],
+        ['1000', 'Indianapolis'],
+      ],
+      rowCount: 2,
+    };
+    printResultTable(multi, 'pretty', io);
+    // Column widened to the longest cell ("1000"), so "1" is right-padded to 4 chars.
+    expect(out()).toContain('1     NY');
+    expect(out()).toContain('2 rows\n');
+  });
+
+  it('pretty mode omits the header rule when there are no headers', () => {
+    const { io, out } = captureIo();
+    printResultTable({ connectionId: 'mongo', headers: [], rows: [], rowCount: 0 }, 'pretty', io);
+    expect(out()).toBe('\n0 rows\n');
+  });
+});
diff --git a/packages/cli/test/mcp/mongo-query-tool.test.ts b/packages/cli/test/mcp/mongo-query-tool.test.ts
new file mode 100644
index 000000000..e0af6023c
--- /dev/null
+++ b/packages/cli/test/mcp/mongo-query-tool.test.ts
@@ -0,0 +1,105 @@
+import { describe, expect, it, vi } from 'vitest';
+import { KtxExpectedError } from '../../src/errors.js';
+import { KtxMongoDbScanConnector } from '../../src/connectors/mongodb/connector.js';
+import { createLocalProjectMcpContextPorts } from '../../src/context/mcp/local-project-ports.js';
+import { runMongoQuery } from '../../src/context/mcp/mongo-query-runner.js';
+import type { KtxLocalProject } from '../../src/context/project/project.js';
+import type { KtxScanConnector } from '../../src/context/scan/types.js';
+
+const MONGO_DOCS = [
+  { _id: 'a1', city: 'Indianapolis' },
+  { _id: 'a2', city: 'Indianapolis' },
+];
+
+function mongoConnector(): KtxMongoDbScanConnector {
+  return new KtxMongoDbScanConnector({
+    connectionId: 'mongo',
+    connection: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] },
+    clientFactory: {
+      create: () => ({
+        listCollections: vi.fn(async () => []),
+        estimatedDocumentCount: vi.fn(async () => 0),
+        find: vi.fn(async () => []),
+        aggregate: vi.fn(async () => MONGO_DOCS),
+        ping: vi.fn(async () => undefined),
+        close: vi.fn(async () => undefined),
+      }),
+    },
+  });
+}
+
+function project(): KtxLocalProject {
+  return {
+    projectDir: '/tmp/ktx',
+    config: {
+      connections: {
+        mongo: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] },
+        warehouse: { driver: 'postgres' },
+      },
+    },
+  } as unknown as KtxLocalProject;
+}
+
+function warehouseConnector(): KtxScanConnector {
+  return {
+    id: 'postgres:warehouse',
+    driver: 'postgres',
+    capabilities: { structuralIntrospection: true } as KtxScanConnector['capabilities'],
+    introspect: vi.fn(),
+    listSchemas: vi.fn(async () => []),
+    listTables: vi.fn(async () => []),
+  } as unknown as KtxScanConnector;
+}
+
+function portsFor(createConnector: (id: string) => KtxScanConnector) {
+  return createLocalProjectMcpContextPorts(project(), {
+    embeddingService: null,
+    localScan: { createConnector } as never,
+  });
+}
+
+describe('mongoQuery MCP port', () => {
+  it('fetches real rows from a mongodb connection', async () => {
+    const ports = portsFor(() => mongoConnector());
+    const result = await ports.mongoQuery!.execute({
+      connectionId: 'mongo',
+      collection: 'business',
+      pipeline: [{ $match: { city: 'Indianapolis' } }],
+      limit: 100,
+    });
+    expect(result.headers).toEqual(['_id', 'city']);
+    expect(result.rowCount).toBe(2);
+  });
+
+  it('rejects a non-mongodb connection id with an expected error naming the wrong driver', async () => {
+    const ports = portsFor((id) => (id === 'mongo' ? mongoConnector() : warehouseConnector()));
+    await expect(
+      ports.mongoQuery!.execute({ connectionId: 'warehouse', collection: 'x', pipeline: [], limit: 10 }),
+    ).rejects.toThrow(/is not a MongoDB connection/);
+    await expect(
+      ports.mongoQuery!.execute({ connectionId: 'warehouse', collection: 'x', pipeline: [], limit: 10 }),
+    ).rejects.toBeInstanceOf(KtxExpectedError);
+  });
+});
+
+describe('runMongoQuery shared helper', () => {
+  it('fetches rows from a mongodb connector', async () => {
+    const result = await runMongoQuery(
+      (id) => (id === 'mongo' ? mongoConnector() : warehouseConnector()),
+      { connectionId: 'mongo', collection: 'business', pipeline: [], limit: 100 },
+    );
+    expect(result.rowCount).toBe(2);
+    expect(result.headers).toEqual(['_id', 'city']);
+  });
+
+  it('throws an expected error naming the wrong driver for a non-mongodb connector', async () => {
+    await expect(
+      runMongoQuery((id) => (id === 'mongo' ? mongoConnector() : warehouseConnector()), {
+        connectionId: 'warehouse',
+        collection: 'x',
+        pipeline: [],
+        limit: 10,
+      }),
+    ).rejects.toThrow(/is not a MongoDB connection/);
+  });
+});
diff --git a/packages/cli/test/mongo-query.test.ts b/packages/cli/test/mongo-query.test.ts
new file mode 100644
index 000000000..5b7d17bbc
--- /dev/null
+++ b/packages/cli/test/mongo-query.test.ts
@@ -0,0 +1,102 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { runKtxMongoQuery } from '../src/mongo-query.js';
+import { KtxMongoDbScanConnector } from '../src/connectors/mongodb/connector.js';
+import type { KtxCliIo } from '../src/cli-runtime.js';
+import type { KtxLocalProject } from '../src/context/project/project.js';
+
+function captureIo(): { io: KtxCliIo; out: () => string; err: () => string } {
+  let out = '';
+  let err = '';
+  const io = {
+    stdout: { write: (s: string) => { out += s; return true; }, isTTY: false },
+    stderr: { write: (s: string) => { err += s; return true; } },
+  } as unknown as KtxCliIo;
+  return { io, out: () => out, err: () => err };
+}
+
+function mongoConnector(): KtxMongoDbScanConnector {
+  return new KtxMongoDbScanConnector({
+    connectionId: 'mongo',
+    connection: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] },
+    clientFactory: {
+      create: () => ({
+        listCollections: vi.fn(async () => []),
+        estimatedDocumentCount: vi.fn(async () => 0),
+        find: vi.fn(async () => []),
+        aggregate: vi.fn(async () => [{ _id: 'a1', city: 'Indianapolis' }, { _id: 'a2', city: 'Indianapolis' }]),
+        ping: vi.fn(async () => undefined),
+        close: vi.fn(async () => undefined),
+      }),
+    },
+  });
+}
+
+const project = {
+  projectDir: '/tmp/ktx',
+  config: { connections: { mongo: { driver: 'mongodb', url: 'mongodb://localhost:27017/app', databases: ['app'] } } },
+} as unknown as KtxLocalProject;
+
+const baseArgs = {
+  projectDir: '/tmp/ktx',
+  connectionId: 'mongo',
+  collection: 'business',
+  pipeline: [{ $match: { city: 'Indianapolis' } }],
+  limit: 100,
+  cliVersion: '0.0.0-test',
+};
+
+describe('runKtxMongoQuery', () => {
+  it('prints rows in plain mode and returns exit code 0', async () => {
+    const { io, out } = captureIo();
+    const code = await runKtxMongoQuery({ ...baseArgs, output: 'plain' }, io, {
+      loadProject: async () => project,
+      createScanConnector: (async () => mongoConnector()) as never,
+    });
+    expect(code).toBe(0);
+    expect(out()).toContain('_id\tcity');
+    expect(out()).toContain('a1\tIndianapolis');
+  });
+
+  it('returns exit code 1 and writes the error message for an unconfigured connection', async () => {
+    const { io, err } = captureIo();
+    const code = await runKtxMongoQuery({ ...baseArgs, connectionId: 'nope' }, io, {
+      loadProject: async () => project,
+      createScanConnector: (async () => mongoConnector()) as never,
+    });
+    expect(code).toBe(1);
+    expect(err().length).toBeGreaterThan(0);
+  });
+});
+
+describe('runKtxMongoQuery telemetry', () => {
+  afterEach(() => {
+    vi.unstubAllEnvs();
+  });
+
+  it('emits mongo_query_completed with outcome ok and the pipeline stage count on success', async () => {
+    vi.stubEnv('KTX_TELEMETRY_DEBUG', '1');
+    vi.stubEnv('CI', '');
+    const { io, err } = captureIo();
+    const code = await runKtxMongoQuery({ ...baseArgs, output: 'plain' }, io, {
+      loadProject: async () => project,
+      createScanConnector: (async () => mongoConnector()) as never,
+    });
+    expect(code).toBe(0);
+    expect(err()).toContain('"event":"mongo_query_completed"');
+    expect(err()).toContain('"outcome":"ok"');
+    expect(err()).toContain('"stageCount":1');
+  });
+
+  it('emits mongo_query_completed with outcome error on failure', async () => {
+    vi.stubEnv('KTX_TELEMETRY_DEBUG', '1');
+    vi.stubEnv('CI', '');
+    const { io, err } = captureIo();
+    const code = await runKtxMongoQuery({ ...baseArgs, connectionId: 'nope' }, io, {
+      loadProject: async () => project,
+      createScanConnector: (async () => mongoConnector()) as never,
+    });
+    expect(code).toBe(1);
+    expect(err()).toContain('"event":"mongo_query_completed"');
+    expect(err()).toContain('"outcome":"error"');
+  });
+});
diff --git a/packages/cli/test/telemetry/events.test.ts b/packages/cli/test/telemetry/events.test.ts
index 033c2def5..1ed5f1f1e 100644
--- a/packages/cli/test/telemetry/events.test.ts
+++ b/packages/cli/test/telemetry/events.test.ts
@@ -31,6 +31,7 @@ describe('telemetry event schemas', () => {
       'sl_validate_completed',
       'sl_query_completed',
       'sql_completed',
+      'mongo_query_completed',
       'wiki_query_completed',
       'mcp_request_completed',
       'daemon_started',
diff --git a/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json b/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json
index 405f02400..0bd22bb3d 100644
--- a/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json
+++ b/python/ktx-daemon/src/ktx_daemon/telemetry/events.schema.json
@@ -144,6 +144,18 @@
         "errorClass"
       ]
     },
+    {
+      "name": "mongo_query_completed",
+      "description": "Emitted after ktx mongo-query completes validation and execution.",
+      "fields": [
+        "driver",
+        "isDemoConnection",
+        "stageCount",
+        "durationMs",
+        "outcome",
+        "errorClass"
+      ]
+    },
     {
       "name": "wiki_query_completed",
       "description": "Emitted after a wiki query completes.",
@@ -1056,6 +1068,77 @@
       ],
       "additionalProperties": false
     },
+    "mongo_query_completed": {
+      "$schema": "https://json-schema.org/draft/2020-12/schema",
+      "type": "object",
+      "properties": {
+        "cliVersion": {
+          "type": "string"
+        },
+        "nodeVersion": {
+          "type": "string"
+        },
+        "osPlatform": {
+          "type": "string"
+        },
+        "osRelease": {
+          "type": "string"
+        },
+        "arch": {
+          "type": "string"
+        },
+        "runtime": {
+          "type": "string",
+          "enum": [
+            "node",
+            "daemon-py"
+          ]
+        },
+        "isCi": {
+          "type": "boolean"
+        },
+        "driver": {
+          "type": "string"
+        },
+        "isDemoConnection": {
+          "type": "boolean"
+        },
+        "stageCount": {
+          "type": "integer",
+          "minimum": 0,
+          "maximum": 9007199254740991
+        },
+        "durationMs": {
+          "type": "number",
+          "minimum": 0
+        },
+        "outcome": {
+          "type": "string",
+          "enum": [
+            "ok",
+            "error"
+          ]
+        },
+        "errorClass": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "cliVersion",
+        "nodeVersion",
+        "osPlatform",
+        "osRelease",
+        "arch",
+        "runtime",
+        "isCi",
+        "driver",
+        "isDemoConnection",
+        "stageCount",
+        "durationMs",
+        "outcome"
+      ],
+      "additionalProperties": false
+    },
     "wiki_query_completed": {
       "$schema": "https://json-schema.org/draft/2020-12/schema",
       "type": "object",
diff --git a/python/ktx-daemon/tests/test_telemetry_schema_sync.py b/python/ktx-daemon/tests/test_telemetry_schema_sync.py
index 0cc822f99..21c076cdb 100644
--- a/python/ktx-daemon/tests/test_telemetry_schema_sync.py
+++ b/python/ktx-daemon/tests/test_telemetry_schema_sync.py
@@ -30,6 +30,7 @@ def test_python_schema_copy_matches_node_schema() -> None:
         "sl_validate_completed",
         "sl_query_completed",
         "sql_completed",
+        "mongo_query_completed",
         "wiki_query_completed",
         "mcp_request_completed",
         "daemon_started",