diff --git a/src/langbot/pkg/provider/runners/n8nsvapi.py b/src/langbot/pkg/provider/runners/n8nsvapi.py index 24ef7f59c..8addfc02e 100644 --- a/src/langbot/pkg/provider/runners/n8nsvapi.py +++ b/src/langbot/pkg/provider/runners/n8nsvapi.py @@ -39,6 +39,9 @@ class N8nServiceAPIRunner(runner.RequestRunner): # 获取输出键名，默认为response self.output_key = self.pipeline_config['ai']['n8n-service-api'].get('output-key', 'response') + self.response_handling = self.pipeline_config['ai']['n8n-service-api'].get('response-handling', 'reply') + if self.response_handling not in {'reply', 'ignore'}: + raise ValueError(f'Invalid n8n response-handling: {self.response_handling}') # 获取认证类型，默认为none self.auth_type = self.pipeline_config['ai']['n8n-service-api'].get('auth-type', 'none') @@ -262,7 +265,11 @@ class N8nServiceAPIRunner(runner.RequestRunner): async with session.post( self.webhook_url, json=payload, headers=headers, auth=auth, timeout=self.timeout ) as response: - if response.status != 200: + if self.response_handling == 'ignore': + status_ok = 200 <= response.status < 300 + else: + status_ok = response.status == 200 + if not status_ok: error_text = ( await httpclient.read_limited( response, @@ -272,6 +279,11 @@ class N8nServiceAPIRunner(runner.RequestRunner): self.ap.logger.error(f'n8n webhook call failed: {response.status}, {error_text}') raise Exception(f'n8n webhook call failed: {response.status}, {error_text}') + if self.response_handling == 'ignore': + response.release() + self.ap.logger.debug('n8n async webhook accepted; response body ignored') + return + async for chunk in self._process_response(response): if is_stream: yield chunk diff --git a/src/langbot/templates/default-pipeline-config.json b/src/langbot/templates/default-pipeline-config.json index 78e2ec958..74009804f 100644 --- a/src/langbot/templates/default-pipeline-config.json +++ b/src/langbot/templates/default-pipeline-config.json @@ -80,7 +80,8 @@ "header-name": "", "header-value": "", "timeout": 120, - "output-key": "response" + "output-key": "response", + "response-handling": "reply" }, "langflow-api": { "base-url": "http://localhost:7860", diff --git a/src/langbot/templates/metadata/pipeline/ai.yaml b/src/langbot/templates/metadata/pipeline/ai.yaml index ccf009941..e16a33f49 100644 --- a/src/langbot/templates/metadata/pipeline/ai.yaml +++ b/src/langbot/templates/metadata/pipeline/ai.yaml @@ -475,6 +475,25 @@ stages: type: string required: false default: 'response' + - name: response-handling + label: + en_US: Webhook Response Handling + zh_Hans: Webhook 响应处理方式 + description: + en_US: Choose whether LangBot forwards the n8n webhook response to the chat user. Ignore mode requires the n8n Webhook node to use Respond Immediately. + zh_Hans: 选择是否将 n8n Webhook 响应转发给聊天用户。忽略模式要求 n8n Webhook 节点使用“立即响应”。 + type: select + required: false + default: 'reply' + options: + - name: reply + label: + en_US: Forward as chat reply + zh_Hans: 转发为聊天回复 + - name: ignore + label: + en_US: Ignore response body (asynchronous workflow) + zh_Hans: 忽略响应正文（异步工作流） - name: coze-api label: en_US: coze API diff --git a/tests/unit_tests/pipeline/test_n8nsvapi.py b/tests/unit_tests/pipeline/test_n8nsvapi.py index 54266aec4..f7ed7edc6 100644 --- a/tests/unit_tests/pipeline/test_n8nsvapi.py +++ b/tests/unit_tests/pipeline/test_n8nsvapi.py @@ -55,7 +55,7 @@ finally: # --------------------------------------------------------------------------- -def make_runner(output_key: str = 'response') -> N8nServiceAPIRunner: +def make_runner(output_key: str = 'response', response_handling: str = 'reply') -> N8nServiceAPIRunner: ap = Mock() ap.logger = Mock() pipeline_config = { @@ -63,6 +63,7 @@ def make_runner(output_key: str = 'response') -> N8nServiceAPIRunner: 'n8n-service-api': { 'webhook-url': 'http://test-n8n/webhook', 'output-key': output_key, + 'response-handling': response_handling, 'auth-type': 'none', } } @@ -287,6 +288,7 @@ def make_http_session_mock(response_bytes: bytes, status: int = 200): """Mock httpclient.get_session() returning a session whose post() yields response_bytes.""" mock_response = make_mock_response([response_bytes], status=status) mock_response.status = status + mock_response.headers = {} mock_cm = AsyncMock() mock_cm.__aenter__ = AsyncMock(return_value=mock_response) @@ -314,6 +316,91 @@ async def test_call_webhook_nonstream_adapter_plain_json(): assert results[0].content == 'result text' +@pytest.mark.asyncio +@pytest.mark.parametrize('status', [200, 201, 202, 204]) +@pytest.mark.parametrize( + 'response_body', + [ + b'{"message":"Workflow was started"}', + b'{"response":"must not be forwarded"}', + b'plain acknowledgement', + ], +) +async def test_call_webhook_ignore_response_body(response_body: bytes, status: int): + """Ignore mode accepts any HTTP 2xx response without emitting chat output.""" + runner = make_runner(response_handling='ignore') + query = make_query(is_stream=False) + http_session = make_http_session_mock(response_body, status=status) + + with patch('langbot.pkg.provider.runners.n8nsvapi.httpclient.get_session', return_value=http_session): + results = [] + async for message in runner._call_webhook(query): + results.append(message) + + assert results == [] + + +@pytest.mark.asyncio +async def test_call_webhook_ignore_releases_without_reading_response_body(): + """Ignore mode returns after the success status without waiting for the body.""" + runner = make_runner(response_handling='ignore') + query = make_query(is_stream=False) + mock_response = make_mock_response([], status=202) + mock_response.headers = {} + mock_response.release = Mock() + + async def fail_if_read(_size): + raise AssertionError('ignore mode must not read the response body') + yield b'' + + mock_response.content.iter_chunked = fail_if_read + mock_cm = AsyncMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_response) + mock_cm.__aexit__ = AsyncMock(return_value=False) + mock_session = Mock() + mock_session.post = Mock(return_value=mock_cm) + + with patch('langbot.pkg.provider.runners.n8nsvapi.httpclient.get_session', return_value=mock_session): + results = [message async for message in runner._call_webhook(query)] + + assert results == [] + mock_response.release.assert_called_once_with() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('status', [201, 202, 204]) +async def test_call_webhook_reply_mode_preserves_http_200_contract(status: int): + """Reply mode remains backward compatible and rejects non-200 statuses.""" + runner = make_runner(response_handling='reply') + query = make_query(is_stream=False) + http_session = make_http_session_mock(b'', status=status) + + with patch('langbot.pkg.provider.runners.n8nsvapi.httpclient.get_session', return_value=http_session): + with pytest.raises(N8nAPIError, match=f'n8n webhook call failed: {status}'): + async for _ in runner._call_webhook(query): + pass + + +@pytest.mark.asyncio +async def test_call_webhook_ignore_mode_preserves_http_error(): + """Ignore mode must not swallow a failed n8n webhook response.""" + runner = make_runner(response_handling='ignore') + query = make_query(is_stream=False) + http_session = make_http_session_mock(b'{"error":"unavailable"}', status=500) + + with patch('langbot.pkg.provider.runners.n8nsvapi.httpclient.get_session', return_value=http_session): + with pytest.raises(N8nAPIError, match='n8n webhook call exception'): + async for _ in runner._call_webhook(query): + pass + + +@pytest.mark.asyncio +async def test_invalid_response_handling_is_rejected(): + """Configuration errors should fail fast instead of silently changing reply behavior.""" + with pytest.raises(ValueError, match='Invalid n8n response-handling'): + make_runner(response_handling='unexpected') + + @pytest.mark.asyncio async def test_call_webhook_stream_adapter_stream_format(): """Stream adapter + stream format → MessageChunks, last is_final.""" diff --git a/web/src/app/home/components/dynamic-form/N8nAuthFieldVisibility.ts b/web/src/app/home/components/dynamic-form/N8nAuthFieldVisibility.ts new file mode 100644 index 000000000..4bae431a5 --- /dev/null +++ b/web/src/app/home/components/dynamic-form/N8nAuthFieldVisibility.ts @@ -0,0 +1,22 @@ +const COMMON_N8N_CONFIG_FIELDS = new Set([ + 'webhook-url', + 'auth-type', + 'timeout', + 'output-key', + 'response-handling', +]); + +export function shouldShowN8nConfigField( + fieldName: string, + authType: string, +): boolean { + if (COMMON_N8N_CONFIG_FIELDS.has(fieldName)) { + return true; + } + + return ( + (authType === 'basic' && fieldName.startsWith('basic-')) || + (authType === 'jwt' && fieldName.startsWith('jwt-')) || + (authType === 'header' && fieldName.startsWith('header-')) + ); +} diff --git a/web/src/app/home/components/dynamic-form/N8nAuthFormComponent.tsx b/web/src/app/home/components/dynamic-form/N8nAuthFormComponent.tsx index 6d80b659b..94770b33e 100644 --- a/web/src/app/home/components/dynamic-form/N8nAuthFormComponent.tsx +++ b/web/src/app/home/components/dynamic-form/N8nAuthFormComponent.tsx @@ -13,6 +13,7 @@ import { import { IDynamicFormItemSchema } from '@/app/infra/entities/form/dynamic'; import DynamicFormItemComponent from '@/app/home/components/dynamic-form/DynamicFormItemComponent'; import { normalizeDynamicFormValuesForSave } from '@/app/home/components/dynamic-form/DynamicFormSaveValues'; +import { shouldShowN8nConfigField } from '@/app/home/components/dynamic-form/N8nAuthFieldVisibility'; import { extractI18nObject } from '@/i18n/I18nProvider'; /** @@ -181,29 +182,9 @@ export default function N8nAuthFormComponent({ }, [form, itemConfigList]); // 根据认证类型过滤表单项 - const filteredConfigList = itemConfigList.filter((config) => { - // 始终显示webhook-url、auth-type、timeout和output-key - if ( - ['webhook-url', 'auth-type', 'timeout', 'output-key'].includes( - config.name, - ) - ) { - return true; - } - - // 根据认证类型显示相应的表单项 - if (authType === 'basic' && config.name.startsWith('basic-')) { - return true; - } - if (authType === 'jwt' && config.name.startsWith('jwt-')) { - return true; - } - if (authType === 'header' && config.name.startsWith('header-')) { - return true; - } - - return false; - }); + const filteredConfigList = itemConfigList.filter((config) => + shouldShowN8nConfigField(config.name, authType), + ); return (
diff --git a/web/tests/unit/n8n-auth-field-visibility.test.mjs b/web/tests/unit/n8n-auth-field-visibility.test.mjs new file mode 100644 index 000000000..7a022d774 --- /dev/null +++ b/web/tests/unit/n8n-auth-field-visibility.test.mjs @@ -0,0 +1,52 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import path from 'node:path'; +import test from 'node:test'; +import ts from 'typescript'; +import { fileURLToPath } from 'node:url'; + +const currentDirectory = path.dirname(fileURLToPath(import.meta.url)); +const sourcePath = path.resolve( + currentDirectory, + '../../src/app/home/components/dynamic-form/N8nAuthFieldVisibility.ts', +); + +function loadVisibilityPolicy() { + const source = fs.readFileSync(sourcePath, 'utf8'); + const compiled = ts.transpileModule(source, { + compilerOptions: { module: ts.ModuleKind.CommonJS }, + }).outputText; + const loadedModule = { exports: {} }; + new Function('require', 'module', 'exports', compiled)( + () => { + throw new Error('N8nAuthFieldVisibility must not have runtime imports'); + }, + loadedModule, + loadedModule.exports, + ); + return loadedModule.exports; +} + +test('shows response handling with the other common n8n fields', () => { + const { shouldShowN8nConfigField } = loadVisibilityPolicy(); + + for (const field of [ + 'webhook-url', + 'auth-type', + 'timeout', + 'output-key', + 'response-handling', + ]) { + assert.equal(shouldShowN8nConfigField(field, 'none'), true, field); + } +}); + +test('shows only fields for the selected n8n authentication method', () => { + const { shouldShowN8nConfigField } = loadVisibilityPolicy(); + + assert.equal(shouldShowN8nConfigField('basic-username', 'basic'), true); + assert.equal(shouldShowN8nConfigField('basic-password', 'jwt'), false); + assert.equal(shouldShowN8nConfigField('jwt-secret', 'jwt'), true); + assert.equal(shouldShowN8nConfigField('header-name', 'header'), true); + assert.equal(shouldShowN8nConfigField('unrelated-field', 'none'), false); +});