feat(n8n-runner): support async response handling (#2487)

* feat(n8n-runner): support async response handling

* fix(n8n-runner): expose response handling in form

* fix(n8n-runner): preserve async response semantics

---------

Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>

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

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",

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

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."""

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-'))
+  );
+}

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 (
     <Form {...form}>

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);
+});