mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-30 19:35:36 +08:00
fix(core): Make MCP executions asynchronous (#27762)
Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>
This commit is contained in:
co-authored by
cubic-dev-ai[bot]
parent
59cec2c9e2
commit
054a2d725a
@@ -7,7 +7,6 @@ import {
|
||||
WEBHOOK_NODE_TYPE,
|
||||
type INode,
|
||||
type IWorkflowExecutionDataProcess,
|
||||
UnexpectedError,
|
||||
} from 'n8n-workflow';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
|
||||
@@ -15,7 +14,6 @@ import { createWorkflow } from './mock.utils';
|
||||
import { WorkflowAccessError } from '../mcp.errors';
|
||||
import { createExecuteWorkflowTool, executeWorkflow } from '../tools/execute-workflow.tool';
|
||||
|
||||
import { ActiveExecutions } from '@/active-executions';
|
||||
import { McpService } from '@/modules/mcp/mcp.service';
|
||||
import { Telemetry } from '@/telemetry';
|
||||
import { WorkflowRunner } from '@/workflow-runner';
|
||||
@@ -24,14 +22,12 @@ import { WorkflowFinderService } from '@/workflows/workflow-finder.service';
|
||||
describe('execute-workflow MCP tool', () => {
|
||||
const user = Object.assign(new User(), { id: 'user-1' });
|
||||
let workflowFinderService: WorkflowFinderService;
|
||||
let activeExecutions: ActiveExecutions;
|
||||
let workflowRunner: WorkflowRunner;
|
||||
let telemetry: Telemetry;
|
||||
let mcpService: McpService;
|
||||
|
||||
beforeEach(() => {
|
||||
workflowFinderService = mockInstance(WorkflowFinderService);
|
||||
activeExecutions = mockInstance(ActiveExecutions);
|
||||
workflowRunner = mockInstance(WorkflowRunner);
|
||||
telemetry = mockInstance(Telemetry, {
|
||||
track: jest.fn(),
|
||||
@@ -46,7 +42,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
const tool = createExecuteWorkflowTool(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
telemetry,
|
||||
mcpService,
|
||||
@@ -56,7 +51,7 @@ describe('execute-workflow MCP tool', () => {
|
||||
expect(tool.config).toBeDefined();
|
||||
expect(typeof tool.config.description).toBe('string');
|
||||
expect(tool.config.description).toBe(
|
||||
'Execute a workflow by ID. Returns execution ID and status. To get the full execution results, use the get_execution tool with the returned execution ID. Before executing always ensure you know the input schema by first using the get_workflow_details tool and consulting workflow description',
|
||||
'Execute a workflow by ID. Returns the execution ID immediately without waiting for completion. Before executing always ensure you know the input schema by first using the get_workflow_details tool and consulting workflow description',
|
||||
);
|
||||
expect(tool.config.inputSchema).toBeDefined();
|
||||
expect(tool.config.outputSchema).toBeDefined();
|
||||
@@ -73,7 +68,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'any-workflow',
|
||||
@@ -93,7 +87,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'unpublished-workflow',
|
||||
@@ -112,15 +105,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('execution-id');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: { runData: {} } },
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'unpublished-workflow',
|
||||
@@ -128,7 +116,7 @@ describe('execute-workflow MCP tool', () => {
|
||||
'manual',
|
||||
);
|
||||
|
||||
expect(result.status).toBe('success');
|
||||
expect(result.status).toBe('started');
|
||||
expect(result.executionId).toBe('execution-id');
|
||||
});
|
||||
|
||||
@@ -150,15 +138,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('execution-id');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: { runData: {} } },
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'manual-trigger-workflow',
|
||||
@@ -166,7 +149,7 @@ describe('execute-workflow MCP tool', () => {
|
||||
'manual',
|
||||
);
|
||||
|
||||
expect(result.status).toBe('success');
|
||||
expect(result.status).toBe('started');
|
||||
expect(result.executionId).toBe('execution-id');
|
||||
expect(workflowRunner.run).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
@@ -199,15 +182,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('execution-id');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: { runData: {} } },
|
||||
});
|
||||
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'manual-workflow-with-pindata',
|
||||
@@ -247,15 +225,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('execution-id');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: { runData: {} } },
|
||||
});
|
||||
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'production-workflow-with-pindata',
|
||||
@@ -293,7 +266,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'unsupported-trigger',
|
||||
@@ -305,7 +277,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'unsupported-trigger',
|
||||
@@ -335,7 +306,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'unsupported-trigger',
|
||||
@@ -367,7 +337,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'disabled-trigger',
|
||||
@@ -395,21 +364,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-123');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: {
|
||||
resultData: {
|
||||
runData: {
|
||||
WebhookNode: [{ data: { main: [[{ json: { result: 'webhook success' } }]] } }],
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'webhook-workflow',
|
||||
@@ -425,7 +383,7 @@ describe('execute-workflow MCP tool', () => {
|
||||
);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: 'success',
|
||||
status: 'started',
|
||||
executionId: 'exec-123',
|
||||
});
|
||||
|
||||
@@ -463,15 +421,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-456');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'webhook-workflow',
|
||||
@@ -518,15 +471,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-789');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'chat-workflow',
|
||||
@@ -537,7 +485,7 @@ describe('execute-workflow MCP tool', () => {
|
||||
);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: 'success',
|
||||
status: 'started',
|
||||
executionId: 'exec-789',
|
||||
});
|
||||
|
||||
@@ -576,15 +524,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-101');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'form-workflow',
|
||||
@@ -599,7 +542,7 @@ describe('execute-workflow MCP tool', () => {
|
||||
);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: 'success',
|
||||
status: 'started',
|
||||
executionId: 'exec-101',
|
||||
});
|
||||
|
||||
@@ -622,8 +565,8 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('execution results handling', () => {
|
||||
test('handles successful execution', async () => {
|
||||
describe('execution start', () => {
|
||||
test('returns started status with execution ID', async () => {
|
||||
const workflow = createWorkflow({
|
||||
activeVersionId: uuid(),
|
||||
nodes: [
|
||||
@@ -640,21 +583,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-success');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: {
|
||||
resultData: {
|
||||
runData: {
|
||||
WebhookNode: [{ data: { main: [[{ json: { output: 'success' } }]] } }],
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'success-workflow',
|
||||
@@ -662,145 +594,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: 'success',
|
||||
status: 'started',
|
||||
executionId: 'exec-success',
|
||||
});
|
||||
});
|
||||
|
||||
test('handles execution with error status', async () => {
|
||||
const workflow = createWorkflow({
|
||||
activeVersionId: uuid(),
|
||||
nodes: [
|
||||
{
|
||||
id: 'node-1',
|
||||
name: 'WebhookNode',
|
||||
type: WEBHOOK_NODE_TYPE,
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
disabled: false,
|
||||
parameters: {},
|
||||
} as INode,
|
||||
],
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-error');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'error',
|
||||
data: {
|
||||
resultData: {
|
||||
error: {
|
||||
message: 'Workflow execution failed',
|
||||
name: 'ExecutionError',
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'error-workflow',
|
||||
undefined,
|
||||
);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: 'error',
|
||||
executionId: 'exec-error',
|
||||
error: 'Workflow execution failed',
|
||||
});
|
||||
});
|
||||
|
||||
test('handles execution with result data error', async () => {
|
||||
const workflow = createWorkflow({
|
||||
activeVersionId: uuid(),
|
||||
nodes: [
|
||||
{
|
||||
id: 'node-1',
|
||||
name: 'WebhookNode',
|
||||
type: WEBHOOK_NODE_TYPE,
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
disabled: false,
|
||||
parameters: {},
|
||||
} as INode,
|
||||
],
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-data-error');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: {
|
||||
resultData: {
|
||||
error: {
|
||||
message: 'Node execution failed',
|
||||
name: 'NodeExecutionError',
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'data-error-workflow',
|
||||
undefined,
|
||||
);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: 'error',
|
||||
executionId: 'exec-data-error',
|
||||
error: 'Node execution failed',
|
||||
});
|
||||
});
|
||||
|
||||
test('handles workflow returning undefined data', async () => {
|
||||
const workflow = createWorkflow({
|
||||
activeVersionId: uuid(),
|
||||
nodes: [
|
||||
{
|
||||
id: 'node-1',
|
||||
name: 'WebhookNode',
|
||||
type: WEBHOOK_NODE_TYPE,
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
disabled: false,
|
||||
parameters: {},
|
||||
} as INode,
|
||||
],
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-no-data');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue(undefined);
|
||||
|
||||
await expect(
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'no-data-workflow',
|
||||
undefined,
|
||||
),
|
||||
).rejects.toThrow(UnexpectedError);
|
||||
|
||||
await expect(
|
||||
executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'no-data-workflow',
|
||||
undefined,
|
||||
),
|
||||
).rejects.toThrow('Workflow did not return any data');
|
||||
expect(result.error).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -822,15 +619,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-no-inputs');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'no-inputs-workflow',
|
||||
@@ -871,15 +663,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-telemetry');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
const tool = createExecuteWorkflowTool(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
telemetry,
|
||||
mcpService,
|
||||
@@ -909,7 +696,7 @@ describe('execute-workflow MCP tool', () => {
|
||||
success: true,
|
||||
data: {
|
||||
executionId: 'exec-telemetry',
|
||||
status: 'success',
|
||||
status: 'started',
|
||||
},
|
||||
},
|
||||
}),
|
||||
@@ -922,7 +709,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
const tool = createExecuteWorkflowTool(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
telemetry,
|
||||
mcpService,
|
||||
@@ -960,7 +746,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
const tool = createExecuteWorkflowTool(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
telemetry,
|
||||
mcpService,
|
||||
@@ -1029,15 +814,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-multi');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'multi-trigger-workflow',
|
||||
@@ -1059,50 +839,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
beforeEach(() => {
|
||||
queueModeMcpService = mockInstance(McpService, {
|
||||
isQueueMode: true,
|
||||
createPendingResponse: jest.fn().mockReturnValue({
|
||||
promise: Promise.resolve({
|
||||
status: 'success',
|
||||
data: { resultData: { runData: {} } },
|
||||
}),
|
||||
resolve: jest.fn(),
|
||||
reject: jest.fn(),
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
test('uses pending response promise in queue mode', async () => {
|
||||
const workflow = createWorkflow({
|
||||
activeVersionId: uuid(),
|
||||
nodes: [
|
||||
{
|
||||
id: 'node-1',
|
||||
name: 'WebhookNode',
|
||||
type: WEBHOOK_NODE_TYPE,
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
disabled: false,
|
||||
parameters: {},
|
||||
} as INode,
|
||||
],
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-queue');
|
||||
|
||||
const result = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
queueModeMcpService,
|
||||
'queue-workflow',
|
||||
undefined,
|
||||
);
|
||||
|
||||
expect(queueModeMcpService.createPendingResponse).toHaveBeenCalledWith('exec-queue');
|
||||
expect(activeExecutions.getPostExecutePromise).not.toHaveBeenCalled();
|
||||
expect(result).toMatchObject({
|
||||
status: 'success',
|
||||
executionId: 'exec-queue',
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1127,7 +863,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
queueModeMcpService,
|
||||
'mcp-meta-workflow',
|
||||
@@ -1159,15 +894,10 @@ describe('execute-workflow MCP tool', () => {
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-regular');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'regular-workflow',
|
||||
@@ -1180,41 +910,6 @@ describe('execute-workflow MCP tool', () => {
|
||||
// determines whether queue mode MCP handling is applied
|
||||
expect(runCall.isMcpExecution).toBe(false);
|
||||
});
|
||||
|
||||
test('uses activeExecutions.getPostExecutePromise in regular mode', async () => {
|
||||
const workflow = createWorkflow({
|
||||
activeVersionId: uuid(),
|
||||
nodes: [
|
||||
{
|
||||
id: 'node-1',
|
||||
name: 'WebhookNode',
|
||||
type: WEBHOOK_NODE_TYPE,
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
disabled: false,
|
||||
parameters: {},
|
||||
} as INode,
|
||||
],
|
||||
});
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(workflow);
|
||||
(workflowRunner.run as jest.Mock).mockResolvedValue('exec-regular-2');
|
||||
(activeExecutions.getPostExecutePromise as jest.Mock).mockResolvedValue({
|
||||
status: 'success',
|
||||
data: { resultData: {} },
|
||||
});
|
||||
|
||||
await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
'regular-workflow-2',
|
||||
undefined,
|
||||
);
|
||||
|
||||
expect(activeExecutions.getPostExecutePromise).toHaveBeenCalledWith('exec-regular-2');
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -696,6 +696,68 @@ describe('get-execution MCP tool', () => {
|
||||
});
|
||||
});
|
||||
|
||||
test('handles execution data with circular references', async () => {
|
||||
const tool = createGetExecutionTool(
|
||||
user,
|
||||
executionRepository,
|
||||
workflowFinderService,
|
||||
telemetry,
|
||||
);
|
||||
|
||||
const mockExecutionData = createEmptyRunExecutionData();
|
||||
// Create circular reference in node output data
|
||||
const circularObj: Record<string, string | Record<string, unknown>> = { value: 'test' };
|
||||
circularObj.self = circularObj;
|
||||
|
||||
mockExecutionData.resultData.runData = {
|
||||
Node1: [
|
||||
{
|
||||
executionIndex: 0,
|
||||
startTime: 0,
|
||||
executionTime: 100,
|
||||
source: [],
|
||||
data: { main: [[{ json: circularObj }]] },
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const mockExecution = {
|
||||
id: 'execution-1',
|
||||
workflowId: 'workflow-1',
|
||||
mode: 'manual',
|
||||
status: 'success',
|
||||
startedAt: new Date('2025-01-01T00:00:00.000Z'),
|
||||
stoppedAt: new Date('2025-01-01T00:01:00.000Z'),
|
||||
retryOf: null,
|
||||
retrySuccessId: null,
|
||||
waitTill: null,
|
||||
data: mockExecutionData,
|
||||
};
|
||||
|
||||
(workflowFinderService.findWorkflowForUser as jest.Mock).mockResolvedValue(mockWorkflow);
|
||||
(executionRepository.findWithUnflattenedData as jest.Mock).mockResolvedValue(mockExecution);
|
||||
|
||||
const result = await tool.handler(
|
||||
{
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
includeData: true,
|
||||
nodeNames: undefined,
|
||||
truncateData: undefined,
|
||||
},
|
||||
{} as never,
|
||||
);
|
||||
|
||||
expect(result.structuredContent).toMatchObject({
|
||||
execution: {
|
||||
id: 'execution-1',
|
||||
status: 'success',
|
||||
},
|
||||
});
|
||||
// Should not return an error — circular refs must be handled gracefully
|
||||
expect(result.structuredContent).not.toHaveProperty('error');
|
||||
});
|
||||
|
||||
test('handles execution with waitTill', async () => {
|
||||
const tool = createGetExecutionTool(
|
||||
user,
|
||||
|
||||
@@ -121,7 +121,6 @@ export class McpService {
|
||||
const executeWorkflowTool = createExecuteWorkflowTool(
|
||||
user,
|
||||
this.workflowFinderService,
|
||||
this.activeExecutions,
|
||||
this.workflowRunner,
|
||||
this.telemetry,
|
||||
this,
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import { Time } from '@n8n/constants';
|
||||
import type { User } from '@n8n/db';
|
||||
import {
|
||||
CHAT_TRIGGER_NODE_TYPE,
|
||||
@@ -21,17 +20,15 @@ import {
|
||||
SUPPORTED_PRODUCTION_MCP_TRIGGERS,
|
||||
USER_CALLED_MCP_TOOL_EVENT,
|
||||
} from '../mcp.constants';
|
||||
import { McpExecutionTimeoutError, WorkflowAccessError } from '../mcp.errors';
|
||||
import { WorkflowAccessError } from '../mcp.errors';
|
||||
import type {
|
||||
ExecuteWorkflowsInputMeta,
|
||||
ToolDefinition,
|
||||
UserCalledMCPToolEventPayload,
|
||||
} from '../mcp.types';
|
||||
import { findMcpSupportedTrigger } from '../mcp.utils';
|
||||
import { waitForExecutionResult, WORKFLOW_EXECUTION_TIMEOUT_MS } from './execution-utils';
|
||||
import { getMcpWorkflow, type FoundWorkflow } from './workflow-validation.utils';
|
||||
|
||||
import type { ActiveExecutions } from '@/active-executions';
|
||||
import type { McpService } from '@/modules/mcp/mcp.service';
|
||||
import type { Telemetry } from '@/telemetry';
|
||||
import type { WorkflowRunner } from '@/workflow-runner';
|
||||
@@ -86,22 +83,19 @@ const inputSchema = z.object({
|
||||
|
||||
type ExecuteWorkflowOutput = {
|
||||
executionId: string | null;
|
||||
status: 'success' | 'error' | 'running' | 'waiting' | 'canceled' | 'crashed' | 'new' | 'unknown';
|
||||
status: 'started' | 'error';
|
||||
error?: string;
|
||||
};
|
||||
|
||||
const outputSchema = {
|
||||
executionId: z.string().nullable(),
|
||||
status: z
|
||||
.enum(['success', 'error', 'running', 'waiting', 'canceled', 'crashed', 'new', 'unknown'])
|
||||
.describe('The status of the execution'),
|
||||
status: z.enum(['started', 'error']).describe('The status of the execution'),
|
||||
error: z.string().optional().describe('Error message if the execution failed'),
|
||||
} satisfies z.ZodRawShape;
|
||||
|
||||
export const createExecuteWorkflowTool = (
|
||||
user: User,
|
||||
workflowFinderService: WorkflowFinderService,
|
||||
activeExecutions: ActiveExecutions,
|
||||
workflowRunner: WorkflowRunner,
|
||||
telemetry: Telemetry,
|
||||
mcpService: McpService,
|
||||
@@ -109,7 +103,7 @@ export const createExecuteWorkflowTool = (
|
||||
name: 'execute_workflow',
|
||||
config: {
|
||||
description:
|
||||
'Execute a workflow by ID. Returns execution ID and status. To get the full execution results, use the get_execution tool with the returned execution ID. Before executing always ensure you know the input schema by first using the get_workflow_details tool and consulting workflow description',
|
||||
'Execute a workflow by ID. Returns the execution ID immediately without waiting for completion. Before executing always ensure you know the input schema by first using the get_workflow_details tool and consulting workflow description',
|
||||
inputSchema: inputSchema.shape,
|
||||
outputSchema,
|
||||
annotations: {
|
||||
@@ -130,7 +124,6 @@ export const createExecuteWorkflowTool = (
|
||||
const output = await executeWorkflow(
|
||||
user,
|
||||
workflowFinderService,
|
||||
activeExecutions,
|
||||
workflowRunner,
|
||||
mcpService,
|
||||
workflowId,
|
||||
@@ -139,15 +132,12 @@ export const createExecuteWorkflowTool = (
|
||||
);
|
||||
|
||||
telemetryPayload.results = {
|
||||
success: output.status === 'success',
|
||||
success: true,
|
||||
data: {
|
||||
executionId: output.executionId,
|
||||
status: output.status,
|
||||
},
|
||||
};
|
||||
if (output.status === 'error' && output.error) {
|
||||
telemetryPayload.results.error = output.error;
|
||||
}
|
||||
telemetry.track(USER_CALLED_MCP_TOOL_EVENT, telemetryPayload);
|
||||
|
||||
return {
|
||||
@@ -156,7 +146,6 @@ export const createExecuteWorkflowTool = (
|
||||
};
|
||||
} catch (er) {
|
||||
const error = ensureError(er);
|
||||
const isTimeout = error instanceof McpExecutionTimeoutError;
|
||||
const isAccessError = error instanceof WorkflowAccessError;
|
||||
|
||||
const errorInfo: Record<string, unknown> = {
|
||||
@@ -173,16 +162,14 @@ export const createExecuteWorkflowTool = (
|
||||
}
|
||||
|
||||
const output: ExecuteWorkflowOutput = {
|
||||
executionId: isTimeout ? error.executionId : null,
|
||||
executionId: null,
|
||||
status: 'error',
|
||||
error: isTimeout
|
||||
? `Workflow execution timed out after ${WORKFLOW_EXECUTION_TIMEOUT_MS * Time.milliseconds.toSeconds} seconds (Enforced MCP timeout)`
|
||||
: (error.message ?? `${error.constructor.name}: (no message)`),
|
||||
error: error.message ?? `${error.constructor.name}: (no message)`,
|
||||
};
|
||||
|
||||
telemetryPayload.results = {
|
||||
success: false,
|
||||
error: isTimeout ? 'Workflow execution timed out' : errorInfo,
|
||||
error: errorInfo,
|
||||
error_reason: isAccessError ? error.reason : undefined,
|
||||
};
|
||||
telemetry.track(USER_CALLED_MCP_TOOL_EVENT, telemetryPayload);
|
||||
@@ -204,7 +191,6 @@ export const createExecuteWorkflowTool = (
|
||||
export const executeWorkflow = async (
|
||||
user: User,
|
||||
workflowFinderService: WorkflowFinderService,
|
||||
activeExecutions: ActiveExecutions,
|
||||
workflowRunner: WorkflowRunner,
|
||||
mcpService: McpService,
|
||||
workflowId: string,
|
||||
@@ -228,15 +214,10 @@ export const executeWorkflow = async (
|
||||
);
|
||||
|
||||
const executionId = await workflowRunner.run(runData);
|
||||
const data = await waitForExecutionResult(executionId, activeExecutions, mcpService);
|
||||
const hasError = data.status === 'error' || data.data.resultData?.error;
|
||||
|
||||
return {
|
||||
executionId,
|
||||
status: hasError ? 'error' : data.status,
|
||||
error: hasError
|
||||
? (data.data.resultData?.error?.message ?? 'Execution completed with errors')
|
||||
: undefined,
|
||||
status: 'started',
|
||||
};
|
||||
};
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ const inputSchema = z.object({
|
||||
.boolean()
|
||||
.optional()
|
||||
.describe(
|
||||
'Whether to include the full execution result data. Defaults to false (metadata only). Set to true to include node inputs/outputs.',
|
||||
'Whether to include the full execution result data. Defaults to false (metadata only). Set to true to include node inputs/outputs. Use `false` to quickly check execution status',
|
||||
),
|
||||
nodeNames: z
|
||||
.array(z.string())
|
||||
@@ -153,7 +153,7 @@ export const createGetExecutionTool = (
|
||||
telemetry.track(USER_CALLED_MCP_TOOL_EVENT, telemetryPayload);
|
||||
|
||||
return {
|
||||
content: [{ type: 'text', text: jsonStringify(output) }],
|
||||
content: [{ type: 'text', text: jsonStringify(output, { replaceCircularRefs: true }) }],
|
||||
structuredContent: output,
|
||||
};
|
||||
} catch (er) {
|
||||
@@ -186,7 +186,7 @@ export const createGetExecutionTool = (
|
||||
telemetry.track(USER_CALLED_MCP_TOOL_EVENT, telemetryPayload);
|
||||
|
||||
return {
|
||||
content: [{ type: 'text', text: jsonStringify(output) }],
|
||||
content: [{ type: 'text', text: jsonStringify(output, { replaceCircularRefs: true }) }],
|
||||
structuredContent: output,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -119,7 +119,7 @@ export interface WorkflowDetailsResult {
|
||||
/** Response from execute_workflow tool */
|
||||
export interface ExecuteWorkflowResult {
|
||||
executionId: string | null;
|
||||
status: 'success' | 'error' | 'running' | 'waiting' | 'canceled' | 'crashed' | 'new' | 'unknown';
|
||||
status: 'started' | 'error';
|
||||
error?: string;
|
||||
}
|
||||
|
||||
|
||||
@@ -301,7 +301,7 @@ test.describe(
|
||||
const { apiKey } = await api.rotateMcpApiKey();
|
||||
const result = await api.mcp.internalMcpExecuteWorkflow(apiKey, workflowId);
|
||||
|
||||
expect(result.status).toBe('success');
|
||||
expect(result.status).toBe('started');
|
||||
expect(result.executionId).toBeTruthy();
|
||||
});
|
||||
|
||||
@@ -343,7 +343,7 @@ test.describe(
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.status).toBe('success');
|
||||
expect(result.status).toBe('started');
|
||||
expect(result.executionId).toBeTruthy();
|
||||
});
|
||||
});
|
||||
@@ -358,9 +358,24 @@ test.describe(
|
||||
const { apiKey } = await api.rotateMcpApiKey();
|
||||
|
||||
const execResult = await api.mcp.internalMcpExecuteWorkflow(apiKey, workflowId);
|
||||
expect(execResult.status).toBe('success');
|
||||
expect(execResult.status).toBe('started');
|
||||
expect(execResult.executionId).toBeTruthy();
|
||||
|
||||
// Poll for execution completion since executions are asynchronous
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
const r = await api.mcp.internalMcpGetExecution(
|
||||
apiKey,
|
||||
workflowId,
|
||||
execResult.executionId!,
|
||||
);
|
||||
return r.execution?.status;
|
||||
},
|
||||
{ timeout: 30_000, intervals: [1_000] },
|
||||
)
|
||||
.toBe('success');
|
||||
|
||||
const result = await api.mcp.internalMcpGetExecution(
|
||||
apiKey,
|
||||
workflowId,
|
||||
@@ -371,7 +386,6 @@ test.describe(
|
||||
expect(result.execution).toBeDefined();
|
||||
expect(result.execution!.id).toBe(execResult.executionId);
|
||||
expect(result.execution!.workflowId).toBe(workflowId);
|
||||
expect(result.execution!.status).toBe('success');
|
||||
expect(result.data).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user