diff --git a/packages/context/src/mcp/context-tools.ts b/packages/context/src/mcp/context-tools.ts index 9f84b586..e49e99a0 100644 --- a/packages/context/src/mcp/context-tools.ts +++ b/packages/context/src/mcp/context-tools.ts @@ -143,6 +143,12 @@ const scanArtifactReadSchema = z.object({ path: z.string().min(1), }); +const sqlExecutionSchema = z.object({ + connectionId: connectionIdSchema, + sql: z.string().min(1), + maxRows: z.number().int().min(1).max(10_000).default(1000).optional(), +}); + export function jsonToolResult(structuredContent: T): KtxMcpToolResult { return { content: [{ type: 'text', text: JSON.stringify(structuredContent, null, 2) }], @@ -361,6 +367,34 @@ export function registerKtxContextTools(deps: RegisterKtxContextToolsDeps): void ); } + if (ports.sqlExecution) { + const sqlExecution = ports.sqlExecution; + registerParsedTool( + server, + 'sql_execution', + { + title: 'SQL Execution', + description: + 'Execute one parser-validated read-only SQL query against a configured KTX connection and return structured rows.', + inputSchema: sqlExecutionSchema.shape, + }, + sqlExecutionSchema, + async (input) => { + try { + return jsonToolResult( + await sqlExecution.execute({ + connectionId: input.connectionId, + sql: input.sql, + maxRows: input.maxRows ?? 1000, + }), + ); + } catch (error) { + return jsonErrorToolResult(error instanceof Error ? error.message : String(error)); + } + }, + ); + } + if (ports.ingest) { const ingest = ports.ingest; registerParsedTool( diff --git a/packages/context/src/mcp/server.test.ts b/packages/context/src/mcp/server.test.ts index 193d8f67..f21c4595 100644 --- a/packages/context/src/mcp/server.test.ts +++ b/packages/context/src/mcp/server.test.ts @@ -11,6 +11,8 @@ import type { KtxMcpContextPorts, KtxScanMcpPort, KtxSemanticLayerMcpPort, + KtxSqlExecutionMcpPort, + KtxSqlExecutionResponse, MemoryCapturePort, } from './types.js'; @@ -64,6 +66,63 @@ describe('createKtxMcpServer', () => { }); }); + it('registers parser-gated sql_execution when the host provides a SQL execution port', async () => { + const fake = makeFakeServer(); + const response: KtxSqlExecutionResponse = { + headers: ['status', 'count'], + headerTypes: ['text', 'bigint'], + rows: [['paid', 42]], + rowCount: 1, + }; + const sqlExecution: KtxSqlExecutionMcpPort = { + execute: vi.fn().mockResolvedValue(response), + }; + + createKtxMcpServer({ + server: fake.server, + userContext: { userId: 'local-user' }, + contextTools: { + sqlExecution, + }, + }); + + expect(fake.tools.map((tool) => tool.name)).toEqual(['sql_execution']); + await expect( + getTool(fake.tools, 'sql_execution').handler({ + connectionId: 'warehouse', + sql: 'select status, count(*) from public.orders group by status', + maxRows: 50, + }), + ).resolves.toEqual({ + content: [ + { + type: 'text', + text: JSON.stringify( + { + headers: ['status', 'count'], + headerTypes: ['text', 'bigint'], + rows: [['paid', 42]], + rowCount: 1, + }, + null, + 2, + ), + }, + ], + structuredContent: { + headers: ['status', 'count'], + headerTypes: ['text', 'bigint'], + rows: [['paid', 42]], + rowCount: 1, + }, + }); + expect(sqlExecution.execute).toHaveBeenCalledWith({ + connectionId: 'warehouse', + sql: 'select status, count(*) from public.orders group by status', + maxRows: 50, + }); + }); + it('registers memory capture tools without host app dependencies', async () => { const fake = makeFakeServer(); const capture: MemoryCapturePort = { diff --git a/packages/context/src/mcp/types.ts b/packages/context/src/mcp/types.ts index f68444b2..df97e3ef 100644 --- a/packages/context/src/mcp/types.ts +++ b/packages/context/src/mcp/types.ts @@ -312,10 +312,22 @@ export interface KtxScanMcpPort { readArtifact?(input: { runId: string; path: string }): Promise; } +export interface KtxSqlExecutionResponse { + headers: string[]; + headerTypes?: string[]; + rows: unknown[][]; + rowCount: number; +} + +export interface KtxSqlExecutionMcpPort { + execute(input: { connectionId: string; sql: string; maxRows: number }): Promise; +} + export interface KtxMcpContextPorts { connections?: KtxConnectionsMcpPort; knowledge?: KtxKnowledgeMcpPort; semanticLayer?: KtxSemanticLayerMcpPort; + sqlExecution?: KtxSqlExecutionMcpPort; ingest?: KtxIngestMcpPort; scan?: KtxScanMcpPort; }