diff --git a/packages/cli/src/context/mcp/local-project-ports.ts b/packages/cli/src/context/mcp/local-project-ports.ts index 684c002c..71aa0456 100644 --- a/packages/cli/src/context/mcp/local-project-ports.ts +++ b/packages/cli/src/context/mcp/local-project-ports.ts @@ -15,6 +15,7 @@ import { KtxDaemonComputeError, type KtxSemanticLayerComputePort } from '../../c import type { KtxLocalProject } from '../../context/project/project.js'; import { createKtxEntityDetailsService } from '../../context/scan/entity-details.js'; import type { LocalScanMcpOptions } from '../../context/scan/local-scan.js'; +import type { KtxScanConnector } from '../../context/scan/types.js'; import { createKtxDiscoverDataService } from '../../context/search/discover.js'; import { sqlAnalysisDialectForDriver } from '../../context/sql-analysis/dialect.js'; import type { SqlAnalysisPort } from '../../context/sql-analysis/ports.js'; @@ -24,7 +25,8 @@ 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 type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxSqlExecutionResponse } from './types.js'; +import type { KtxMongoDbScanConnector } from '../../connectors/mongodb/connector.js'; +import type { KtxMcpContextPorts, KtxMcpProgressCallback, KtxMongoQueryResponse, KtxSqlExecutionResponse } from './types.js'; interface CreateLocalProjectMcpContextPortsOptions { semanticLayerCompute?: KtxSemanticLayerComputePort; @@ -54,6 +56,35 @@ 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 executeMongoQuery( + createConnector: (connectionId: string) => Promise | KtxScanConnector, + input: { connectionId: string; collection: string; database?: string; pipeline: Record[]; limit: number }, +): Promise { + const connectionId = assertSafeConnectionId(input.connectionId); + let connector: KtxScanConnector | null = null; + try { + connector = await createConnector(connectionId); + if (connector.driver !== 'mongodb') { + throw new KtxExpectedError( + `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, + ); + } + const result = await (connector as unknown as KtxMongoDbScanConnector).executeQuery( + { connectionId, collection: input.collection, database: input.database, pipeline: input.pipeline, limit: input.limit }, + { runId: 'mcp-mongo-query' }, + ); + return { headers: result.headers, rows: result.rows, rowCount: result.rowCount }; + } finally { + await connector?.cleanup?.(); + } +} + async function executeValidatedReadOnlySql( project: KtxLocalProject, options: CreateLocalProjectMcpContextPortsOptions, @@ -239,5 +270,18 @@ export function createLocalProjectMcpContextPorts( }; } + const mongoCreateConnector = options.localScan?.createConnector; + if (mongoCreateConnector && projectHasMongoConnection(project)) { + ports.mongoQuery = { + async execute(input) { + try { + return await executeMongoQuery(mongoCreateConnector, input); + } catch (error) { + throwClassifiedQueryError(error); + } + }, + }; + } + return ports; } 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 00000000..4d316baf --- /dev/null +++ b/packages/cli/test/mcp/mongo-query-tool.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it, vi } from 'vitest'; +import { KtxQueryError } from '../../src/errors.js'; +import { KtxMongoDbScanConnector } from '../../src/connectors/mongodb/connector.js'; +import { createLocalProjectMcpContextPorts } from '../../src/context/mcp/local-project-ports.js'; +import type { KtxLocalProject } from '../../src/context/project/project.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 portsFor(createConnector: (id: string) => KtxMongoDbScanConnector) { + 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 as an expected query error', async () => { + const ports = portsFor(() => mongoConnector()); + await expect( + ports.mongoQuery!.execute({ connectionId: 'warehouse', collection: 'x', pipeline: [], limit: 10 }), + ).rejects.toBeInstanceOf(KtxQueryError); + }); +});