fix(Call n8n Workflow Tool Node): Return all items from subexecution (#13393)

This commit is contained in:
Mutasem Aldmour 2025-03-03 15:33:02 +01:00 committed by GitHub
parent b7f7121cb8
commit d9e3cfe13f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 114 additions and 15 deletions

View file

@ -28,7 +28,7 @@ export class ToolWorkflow extends VersionedNodeType {
], ],
}, },
}, },
defaultVersion: 2, defaultVersion: 2.1,
}; };
const nodeVersions: IVersionedNodeType['nodeVersions'] = { const nodeVersions: IVersionedNodeType['nodeVersions'] = {
@ -37,6 +37,7 @@ export class ToolWorkflow extends VersionedNodeType {
1.2: new ToolWorkflowV1(baseDescription), 1.2: new ToolWorkflowV1(baseDescription),
1.3: new ToolWorkflowV1(baseDescription), 1.3: new ToolWorkflowV1(baseDescription),
2: new ToolWorkflowV2(baseDescription), 2: new ToolWorkflowV2(baseDescription),
2.1: new ToolWorkflowV2(baseDescription),
}; };
super(nodeVersions, baseDescription); super(nodeVersions, baseDescription);
} }

View file

@ -25,7 +25,9 @@ export class ToolWorkflowV2 implements INodeType {
}; };
async supplyData(this: ISupplyDataFunctions, itemIndex: number): Promise<SupplyData> { async supplyData(this: ISupplyDataFunctions, itemIndex: number): Promise<SupplyData> {
const workflowToolService = new WorkflowToolService(this); const returnAllItems = this.getNode().typeVersion > 2;
const workflowToolService = new WorkflowToolService(this, { returnAllItems });
const name = this.getNodeParameter('name', itemIndex) as string; const name = this.getNodeParameter('name', itemIndex) as string;
const description = this.getNodeParameter('description', itemIndex) as string; const description = this.getNodeParameter('description', itemIndex) as string;

View file

@ -187,6 +187,66 @@ describe('WorkflowTool::WorkflowToolService', () => {
expect(result.subExecutionId).toBe('test-execution'); expect(result.subExecutionId).toBe('test-execution');
}); });
it('should successfully execute workflow and return first item of many', async () => {
const workflowInfo = { id: 'test-workflow' };
const items: INodeExecutionData[] = [];
const workflowProxyMock = {
$execution: { id: 'exec-id' },
$workflow: { id: 'workflow-id' },
} as unknown as IWorkflowDataProxyData;
const TEST_RESPONSE_1 = { msg: 'test response 1' };
const TEST_RESPONSE_2 = { msg: 'test response 2' };
const mockResponse: ExecuteWorkflowData = {
data: [[{ json: TEST_RESPONSE_1 }, { json: TEST_RESPONSE_2 }]],
executionId: 'test-execution',
};
jest.spyOn(context, 'executeWorkflow').mockResolvedValueOnce(mockResponse);
const result = await service['executeSubWorkflow'](
context,
workflowInfo,
items,
workflowProxyMock,
);
expect(result.response).toBe(TEST_RESPONSE_1);
expect(result.subExecutionId).toBe('test-execution');
});
it('should successfully execute workflow and return all items', async () => {
const serviceWithReturnAllItems = new WorkflowToolService(context, { returnAllItems: true });
const workflowInfo = { id: 'test-workflow' };
const items: INodeExecutionData[] = [];
const workflowProxyMock = {
$execution: { id: 'exec-id' },
$workflow: { id: 'workflow-id' },
} as unknown as IWorkflowDataProxyData;
const TEST_RESPONSE_1 = { msg: 'test response 1' };
const TEST_RESPONSE_2 = { msg: 'test response 2' };
const mockResponse: ExecuteWorkflowData = {
data: [[{ json: TEST_RESPONSE_1 }, { json: TEST_RESPONSE_2 }]],
executionId: 'test-execution',
};
jest.spyOn(context, 'executeWorkflow').mockResolvedValueOnce(mockResponse);
const result = await serviceWithReturnAllItems['executeSubWorkflow'](
context,
workflowInfo,
items,
workflowProxyMock,
undefined,
);
expect(result.response).toEqual([{ json: TEST_RESPONSE_1 }, { json: TEST_RESPONSE_2 }]);
expect(result.subExecutionId).toBe('test-execution');
});
it('should throw error when workflow execution fails', async () => { it('should throw error when workflow execution fails', async () => {
jest.spyOn(context, 'executeWorkflow').mockRejectedValueOnce(new Error('Execution failed')); jest.spyOn(context, 'executeWorkflow').mockRejectedValueOnce(new Error('Execution failed'));

View file

@ -1,7 +1,6 @@
import type { CallbackManagerForToolRun } from '@langchain/core/callbacks/manager'; import type { CallbackManagerForToolRun } from '@langchain/core/callbacks/manager';
import { DynamicStructuredTool, DynamicTool } from '@langchain/core/tools'; import { DynamicStructuredTool, DynamicTool } from '@langchain/core/tools';
import get from 'lodash/get'; import { isArray, isObject } from 'lodash';
import isObject from 'lodash/isObject';
import type { SetField, SetNodeOptions } from 'n8n-nodes-base/dist/nodes/Set/v2/helpers/interfaces'; import type { SetField, SetNodeOptions } from 'n8n-nodes-base/dist/nodes/Set/v2/helpers/interfaces';
import * as manual from 'n8n-nodes-base/dist/nodes/Set/v2/manual.mode'; import * as manual from 'n8n-nodes-base/dist/nodes/Set/v2/manual.mode';
import { getCurrentWorkflowInputData } from 'n8n-nodes-base/dist/utils/workflowInputsResourceMapping/GenericFunctions'; import { getCurrentWorkflowInputData } from 'n8n-nodes-base/dist/utils/workflowInputsResourceMapping/GenericFunctions';
@ -29,6 +28,10 @@ import {
} from 'n8n-workflow'; } from 'n8n-workflow';
import { z } from 'zod'; import { z } from 'zod';
function isNodeExecutionData(data: unknown): data is INodeExecutionData[] {
return isArray(data) && Boolean(data.length) && isObject(data[0]) && 'json' in data[0];
}
/** /**
Main class for creating the Workflow tool Main class for creating the Workflow tool
Processes the node parameters and creates AI Agent tool capable of executing n8n workflows Processes the node parameters and creates AI Agent tool capable of executing n8n workflows
@ -43,10 +46,16 @@ export class WorkflowToolService {
// Sub-workflow execution id, will be set after the sub-workflow is executed // Sub-workflow execution id, will be set after the sub-workflow is executed
private subExecutionId: string | undefined; private subExecutionId: string | undefined;
constructor(private baseContext: ISupplyDataFunctions) { private returnAllItems: boolean = false;
constructor(
private baseContext: ISupplyDataFunctions,
options?: { returnAllItems: boolean },
) {
const subWorkflowInputs = this.baseContext.getNode().parameters const subWorkflowInputs = this.baseContext.getNode().parameters
.workflowInputs as ResourceMapperValue; .workflowInputs as ResourceMapperValue;
this.useSchema = (subWorkflowInputs?.schema ?? []).length > 0; this.useSchema = (subWorkflowInputs?.schema ?? []).length > 0;
this.returnAllItems = options?.returnAllItems ?? false;
} }
// Creates the tool based on the provided parameters // Creates the tool based on the provided parameters
@ -65,7 +74,7 @@ export class WorkflowToolService {
const toolHandler = async ( const toolHandler = async (
query: string | IDataObject, query: string | IDataObject,
runManager?: CallbackManagerForToolRun, runManager?: CallbackManagerForToolRun,
): Promise<string> => { ): Promise<IDataObject | IDataObject[] | string> => {
const localRunIndex = runIndex++; const localRunIndex = runIndex++;
// We need to clone the context here to handle runIndex correctly // We need to clone the context here to handle runIndex correctly
// Otherwise the runIndex will be shared between different executions // Otherwise the runIndex will be shared between different executions
@ -74,10 +83,23 @@ export class WorkflowToolService {
runIndex: localRunIndex, runIndex: localRunIndex,
inputData: [[{ json: { query } }]], inputData: [[{ json: { query } }]],
}); });
try { try {
const response = await this.runFunction(context, query, itemIndex, runManager); const response = await this.runFunction(context, query, itemIndex, runManager);
const processedResponse = this.handleToolResponse(response); const processedResponse = this.handleToolResponse(response);
let responseData: INodeExecutionData[];
if (isNodeExecutionData(response)) {
responseData = response;
} else {
const reParsedData = jsonParse<IDataObject>(processedResponse, {
fallbackValue: { response: processedResponse },
});
responseData = [{ json: reParsedData }];
}
// Once the sub-workflow is executed, add the output data to the context // Once the sub-workflow is executed, add the output data to the context
// This will be used to link the sub-workflow execution in the parent workflow // This will be used to link the sub-workflow execution in the parent workflow
let metadata: ITaskMetadata | undefined; let metadata: ITaskMetadata | undefined;
@ -89,13 +111,11 @@ export class WorkflowToolService {
}, },
}; };
} }
const json = jsonParse<IDataObject>(processedResponse, {
fallbackValue: { response: processedResponse },
});
void context.addOutputData( void context.addOutputData(
NodeConnectionType.AiTool, NodeConnectionType.AiTool,
localRunIndex, localRunIndex,
[[{ json }]], [responseData],
metadata, metadata,
); );
@ -126,6 +146,14 @@ export class WorkflowToolService {
return response.toString(); return response.toString();
} }
if (isNodeExecutionData(response)) {
return JSON.stringify(
response.map((item) => item.json),
null,
2,
);
}
if (isObject(response)) { if (isObject(response)) {
return JSON.stringify(response, null, 2); return JSON.stringify(response, null, 2);
} }
@ -148,7 +176,7 @@ export class WorkflowToolService {
items: INodeExecutionData[], items: INodeExecutionData[],
workflowProxy: IWorkflowDataProxyData, workflowProxy: IWorkflowDataProxyData,
runManager?: CallbackManagerForToolRun, runManager?: CallbackManagerForToolRun,
): Promise<{ response: string; subExecutionId: string }> { ): Promise<{ response: string | IDataObject | INodeExecutionData[]; subExecutionId: string }> {
let receivedData: ExecuteWorkflowData; let receivedData: ExecuteWorkflowData;
try { try {
receivedData = await context.executeWorkflow(workflowInfo, items, runManager?.getChild(), { receivedData = await context.executeWorkflow(workflowInfo, items, runManager?.getChild(), {
@ -163,7 +191,12 @@ export class WorkflowToolService {
throw new NodeOperationError(context.getNode(), error as Error); throw new NodeOperationError(context.getNode(), error as Error);
} }
const response: string | undefined = get(receivedData, 'data[0][0].json') as string | undefined; let response: IDataObject | INodeExecutionData[] | undefined;
if (this.returnAllItems) {
response = receivedData?.data?.[0]?.length ? receivedData.data[0] : undefined;
} else {
response = receivedData?.data?.[0]?.[0]?.json;
}
if (response === undefined) { if (response === undefined) {
throw new NodeOperationError( throw new NodeOperationError(
context.getNode(), context.getNode(),
@ -183,7 +216,7 @@ export class WorkflowToolService {
query: string | IDataObject, query: string | IDataObject,
itemIndex: number, itemIndex: number,
runManager?: CallbackManagerForToolRun, runManager?: CallbackManagerForToolRun,
): Promise<string> { ): Promise<string | IDataObject | INodeExecutionData[]> {
const source = context.getNodeParameter('source', itemIndex) as string; const source = context.getNodeParameter('source', itemIndex) as string;
const workflowProxy = context.getWorkflowDataProxy(0); const workflowProxy = context.getWorkflowDataProxy(0);
@ -304,7 +337,10 @@ export class WorkflowToolService {
private async createStructuredTool( private async createStructuredTool(
name: string, name: string,
description: string, description: string,
func: (query: string | IDataObject, runManager?: CallbackManagerForToolRun) => Promise<string>, func: (
query: string | IDataObject,
runManager?: CallbackManagerForToolRun,
) => Promise<string | IDataObject | IDataObject[]>,
): Promise<DynamicStructuredTool | DynamicTool> { ): Promise<DynamicStructuredTool | DynamicTool> {
const collectedArguments = await this.extractFromAIParameters(); const collectedArguments = await this.extractFromAIParameters();

View file

@ -12,7 +12,7 @@ export const versionDescription: INodeTypeDescription = {
defaults: { defaults: {
name: 'Call n8n Workflow Tool', name: 'Call n8n Workflow Tool',
}, },
version: [2], version: [2, 2.1],
inputs: [], inputs: [],
outputs: [NodeConnectionType.AiTool], outputs: [NodeConnectionType.AiTool],
outputNames: ['Tool'], outputNames: ['Tool'],