end

pyproject.toml

@@ -70,7 +70,7 @@ dependencies = [
     "chromadb>=1.0.0,<2.0.0",
     "qdrant-client (>=1.15.1,<2.0.0)",
     "pyseekdb==1.1.0.post3",
-    "langbot-plugin==0.3.11",
+    "langbot-plugin @ file:///home/typer/Desktop/langbot-plugin-sdk",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0",
     "matrix-nio>=0.25.2",

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -384,7 +384,8 @@ class RuntimePipeline:
 
         finally:
             self.ap.logger.debug(f'Query {query.query_id} processed')
-            del self.ap.query_pool.cached_queries[query.query_id]
+            # Use pop with default to avoid KeyError if query was never cached
+            self.ap.query_pool.cached_queries.pop(query.query_id, None)
 
 
 class PipelineManager:

src/langbot/pkg/workflow/nodes/call_pipeline.py

@@ -188,11 +188,10 @@ class _WorkflowPipelineCaptureAdapter(abstract_platform_adapter.AbstractMessageP
         arbitrary_types_allowed = True
 
     responses: list[dict[str, Any]] = []
-    context: ExecutionContext = pydantic.Field(exclude=True)
+    context: Optional[ExecutionContext] = pydantic.Field(default=None, exclude=True)
 
     def __init__(self, context: ExecutionContext):
-        super().__init__(config={}, logger=_NoOpEventLogger())
+        super().__init__(config={}, logger=_NoOpEventLogger(), context=context)
-        self.context = context
         self.responses = []
 
     async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):

src/langbot/pkg/workflow/nodes/code_executor.py

@@ -5,13 +5,94 @@ Node metadata is loaded from: ../../templates/metadata/nodes/code_executor.yaml
 
 from __future__ import annotations
 
-import json
+import ast
-import re
+import io
+import logging
+import sys
+import threading
 from typing import Any
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
 from ..node import WorkflowNode, workflow_node
 
+logger = logging.getLogger(__name__)
+
+# 危险的内置函数和模块黑名单
+_DANGEROUS_BUILTINS = {
+    '__import__', 'eval', 'exec', 'compile', 'open', 'file',
+    'input', 'exit', 'quit', 'globals', 'locals', 'vars',
+    'dir', 'help', 'breakpoint',
+}
+
+# 允许的安全内置函数
+_SAFE_BUILTINS = {
+    'abs': abs, 'all': all, 'any': any, 'bin': bin, 'bool': bool,
+    'bytearray': bytearray, 'bytes': bytes, 'callable': callable,
+    'chr': chr, 'complex': complex, 'dict': dict, 'divmod': divmod,
+    'enumerate': enumerate, 'filter': filter, 'float': float,
+    'format': format, 'frozenset': frozenset, 'hash': hash,
+    'hex': hex, 'int': int, 'isinstance': isinstance, 'issubclass': issubclass,
+    'iter': iter, 'len': len, 'list': list, 'map': map, 'max': max,
+    'min': min, 'next': next, 'object': object, 'oct': oct, 'ord': ord,
+    'pow': pow, 'print': print, 'range': range, 'repr': repr,
+    'reversed': reversed, 'round': round, 'set': set, 'slice': slice,
+    'sorted': sorted, 'str': str, 'sum': sum, 'tuple': tuple,
+    'type': type, 'zip': zip,
+}
+
+
+def _check_code_safety(code: str) -> list[str]:
+    """检查代码中是否包含危险操作"""
+    warnings = []
+    try:
+        tree = ast.parse(code)
+        for node in ast.walk(tree):
+            # 检查 import 语句
+            if isinstance(node, (ast.Import, ast.ImportFrom)):
+                warnings.append('Import statements are not allowed')
+            # 检查危险函数调用
+            if isinstance(node, ast.Call):
+                if isinstance(node.func, ast.Name) and node.func.id in _DANGEROUS_BUILTINS:
+                    warnings.append(f'Dangerous function call: {node.func.id}')
+                # 检查 __import__ 通过 getattr 调用
+                if isinstance(node.func, ast.Attribute):
+                    if node.func.attr in ('__import__', 'eval', 'exec', 'open', 'file'):
+                        warnings.append(f'Dangerous attribute access: {node.func.attr}')
+    except SyntaxError as e:
+        warnings.append(f'Syntax error in code: {e}')
+    return warnings
+
+
+class _ExecutionTimeoutError(Exception):
+    """执行超时错误"""
+    pass
+
+
+def _run_with_timeout(func, timeout: float = 10.0):
+    """带超时限制的函数执行"""
+    result = [None]
+    error = [None]
+
+    def _target():
+        try:
+            result[0] = func()
+        except Exception as e:
+            error[0] = e
+
+    thread = threading.Thread(target=_target)
+    thread.daemon = True
+    thread.start()
+    thread.join(timeout)
+
+    if thread.is_alive():
+        raise _ExecutionTimeoutError(f'Code execution timed out after {timeout} seconds')
+
+    if error[0]:
+        raise error[0]
+
+    return result[0]
+
+
 @workflow_node('code_executor')
 class CodeExecutorNode(WorkflowNode):
     """Code executor node - run Python or JavaScript code"""
@@ -21,61 +102,55 @@ class CodeExecutorNode(WorkflowNode):
     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
         code = self.get_config('code', '')
         language = self.get_config('language', 'python')
+        timeout = self.get_config('timeout', 10)
+
+        # 限制最大超时时间
+        timeout = min(max(timeout, 1), 30)
+
+        if not code:
+            return {'output': None, 'console': '', 'error': 'No code provided'}
 
         if language == 'python':
-            return await self._execute_python(code, inputs, context)
+            return await self._execute_python(code, inputs, context, timeout)
         else:
             return await self._execute_javascript(code, inputs, context)
 
-    async def _execute_python(self, code: str, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
+    async def _execute_python(self, code: str, inputs: dict[str, Any], context: ExecutionContext, timeout: float) -> dict[str, Any]:
-        import io
+        # 安全检查
-        import sys
+        warnings = _check_code_safety(code)
+        if warnings:
+            logger.warning('Code safety warnings: %s', warnings)
+            return {'output': None, 'console': '', 'error': '; '.join(warnings)}
 
         stdout_capture = io.StringIO()
         old_stdout = sys.stdout
 
-        try:
+        def _exec_code():
+            nonlocal stdout_capture
             sys.stdout = stdout_capture
-
+            try:
-            restricted_globals = {
+                # 使用更安全的执行方式
-                '__builtins__': {
+                compiled = compile(code, '<workflow>', 'exec')
-                    'len': len,
+                safe_globals = {
-                    'str': str,
+                    '__builtins__': _SAFE_BUILTINS,
-                    'int': int,
+                    '__name__': '__workflow_sandbox__',
-                    'float': float,
-                    'bool': bool,
-                    'list': list,
-                    'dict': dict,
-                    'set': set,
-                    'tuple': tuple,
-                    'range': range,
-                    'enumerate': enumerate,
-                    'zip': zip,
-                    'map': map,
-                    'filter': filter,
-                    'sorted': sorted,
-                    'reversed': reversed,
-                    'sum': sum,
-                    'min': min,
-                    'max': max,
-                    'abs': abs,
-                    'round': round,
-                    'print': print,
-                    'isinstance': isinstance,
-                    'type': type,
-                    'hasattr': hasattr,
-                    'getattr': getattr,
-                    'json': json,
-                    're': re,
                 }
-            }
+                local_vars = {'inputs': inputs, 'output': None}
+                exec(compiled, safe_globals, local_vars)
+                return local_vars.get('output')
+            finally:
+                sys.stdout = old_stdout
 
-            local_vars = {'inputs': inputs, 'output': None}
+        try:
-            exec(code, restricted_globals, local_vars)
+            output = _run_with_timeout(_exec_code, timeout)
-
+            console_output = stdout_capture.getvalue()
-            return {'output': local_vars.get('output'), 'console': stdout_capture.getvalue()}
+            return {'output': output, 'console': console_output, 'error': None}
-        finally:
+        except _ExecutionTimeoutError as e:
-            sys.stdout = old_stdout
+            logger.error('Code execution timeout: %s', e)
+            return {'output': None, 'console': stdout_capture.getvalue(), 'error': str(e)}
+        except Exception as e:
+            logger.error('Code execution error: %s', e)
+            return {'output': None, 'console': stdout_capture.getvalue(), 'error': f'{type(e).__name__}: {e}'}
 
     async def _execute_javascript(self, code: str, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        return {'output': f'[JS execution not implemented: {code[:50]}...]', 'console': ''}
+        return {'output': None, 'console': '', 'error': 'JavaScript execution is not implemented'}

src/langbot/pkg/workflow/nodes/condition.py

@@ -5,12 +5,51 @@ Node metadata is loaded from: ../../templates/metadata/nodes/condition.yaml
 
 from __future__ import annotations
 
+import logging
+import re
+import signal
 from typing import Any
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
 from ..node import WorkflowNode, workflow_node
 from ..safe_eval import safe_eval_with_vars
 
+logger = logging.getLogger(__name__)
+
+# 正则表达式超时限制（秒）
+_REGEX_TIMEOUT = 2
+
+
+class _RegexTimeoutError(Exception):
+    """正则表达式超时错误"""
+    pass
+
+
+def _handle_timeout(signum, frame):
+    """超时信号处理"""
+    raise _RegexTimeoutError('Regex match timed out')
+
+
+def _safe_regex_match(pattern: str, text: str) -> tuple[bool, str]:
+    """安全地执行正则表达式匹配，带有超时限制"""
+    # 设置超时信号
+    old_handler = signal.signal(signal.SIGALRM, _handle_timeout)
+    signal.setitimer(signal.ITIMER_REAL, _REGEX_TIMEOUT)
+
+    try:
+        result = bool(re.match(pattern, str(text)))
+        return result, ''
+    except _RegexTimeoutError:
+        logger.warning('Regex match timed out for pattern: %s', pattern[:50])
+        return False, 'Regex match timed out'
+    except re.error as e:
+        logger.warning('Invalid regex pattern: %s', e)
+        return False, f'Invalid regex: {e}'
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, old_handler)
+
+
 @workflow_node('condition')
 class ConditionNode(WorkflowNode):
     """Condition node - branch based on condition"""
@@ -35,11 +74,11 @@ class ConditionNode(WorkflowNode):
         elif condition_type == 'empty':
             result = not bool(input_data)
         elif condition_type == 'regex':
-            import re
-
             left = self.get_config('left_value', '')
             pattern = self.get_config('right_value', '')
-            result = bool(re.match(pattern, str(left)))
+            result, error = _safe_regex_match(pattern, left)
+            if error:
+                return {'true': None, 'false': input_data, 'error': error}
 
         if result:
             return {'true': input_data, 'false': None}
@@ -50,7 +89,8 @@ class ConditionNode(WorkflowNode):
         try:
             local_vars = {'input': data, 'data': data, 'variables': context.variables}
             return bool(safe_eval_with_vars(expression, local_vars))
-        except Exception:
+        except Exception as e:
+            logger.warning('Expression evaluation error: %s', e)
             return False
 
     async def _evaluate_comparison(self, data: Any, context: ExecutionContext) -> bool:

src/langbot/pkg/workflow/nodes/coze_bot.py

@@ -23,12 +23,15 @@ class CozeBotNode(WorkflowNode):
         query = inputs.get('query', '')
         conversation_id = inputs.get('conversation_id')
 
+        # Safe API key truncation
+        masked_key = f'{api_key[:4]}...{api_key[-4:]}' if len(api_key) > 8 else '***' if api_key else ''
+
         return {
             'answer': '',
             'conversation_id': conversation_id,
             'success': False,
             '_debug': {
-                'api_key': api_key[:8] + '...' if api_key else '',
+                'api_key': masked_key,
                 'bot_id': bot_id,
                 'api_base': api_base,
                 'query': query,

src/langbot/pkg/workflow/nodes/dify_knowledge_query.py

@@ -22,12 +22,15 @@ class DifyKnowledgeQueryNode(WorkflowNode):
         dataset_id = self.get_config('dataset_id', '')
         query = inputs.get('query', '')
 
+        # Safe API key truncation
+        masked_key = f'{api_key[:4]}...{api_key[-4:]}' if len(api_key) > 8 else '***' if api_key else ''
+
         return {
             'results': [],
             'success': False,
             '_debug': {
                 'base_url': base_url,
-                'api_key': api_key[:8] + '...' if api_key else '',
+                'api_key': masked_key,
                 'dataset_id': dataset_id,
                 'query': query,
             },

src/langbot/pkg/workflow/nodes/dify_workflow.py

@@ -23,13 +23,16 @@ class DifyWorkflowNode(WorkflowNode):
         query = inputs.get('query', '')
         conversation_id = inputs.get('conversation_id')
 
+        # Safe API key truncation
+        masked_key = f'{api_key[:4]}...{api_key[-4:]}' if len(api_key) > 8 else '***' if api_key else ''
+
         return {
             'answer': '',
             'conversation_id': conversation_id,
             'success': False,
             '_debug': {
                 'base_url': base_url,
-                'api_key': api_key[:8] + '...' if api_key else '',
+                'api_key': masked_key,
                 'app_type': app_type,
                 'query': query,
             },

src/langbot/pkg/workflow/nodes/event_trigger.py

@@ -5,6 +5,7 @@ Node metadata is loaded from: ../../templates/metadata/nodes/event_trigger.yaml
 
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
@@ -17,9 +18,8 @@ class EventTriggerNode(WorkflowNode):
     category = 'trigger'
 
     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        from datetime import datetime
+        # Safe access to trigger_data which may be None
-
+        trigger_data = context.trigger_data or {}
-        trigger_data = context.trigger_data
 
         return {
             'event_type': trigger_data.get('event_type', ''),

src/langbot/pkg/workflow/nodes/http_request.py

@@ -5,11 +5,73 @@ Node metadata is loaded from: ../../templates/metadata/nodes/http_request.yaml
 
 from __future__ import annotations
 
+import ipaddress
+import logging
+import re
 from typing import Any
+from urllib.parse import urlparse
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
 from ..node import WorkflowNode, workflow_node
 
+logger = logging.getLogger(__name__)
+
+# 内网地址黑名单
+_PRIVATE_NETWORKS = [
+    ipaddress.ip_network('10.0.0.0/8'),
+    ipaddress.ip_network('172.16.0.0/12'),
+    ipaddress.ip_network('192.168.0.0/16'),
+    ipaddress.ip_network('127.0.0.0/8'),
+    ipaddress.ip_network('169.254.0.0/16'),
+    ipaddress.ip_network('0.0.0.0/8'),
+    ipaddress.ip_network('::1/128'),
+    ipaddress.ip_network('fc00::/7'),
+    ipaddress.ip_network('fe80::/10'),
+]
+
+# 危险协议
+_DANGEROUS_SCHEMES = {'file', 'gopher', 'dict', 'ftp', 'telnet'}
+
+
+def _is_safe_url(url: str) -> tuple[bool, str]:
+    """检查 URL 是否安全（非内网地址）"""
+    try:
+        parsed = urlparse(url)
+    except Exception as e:
+        return False, f'Invalid URL: {e}'
+
+    # 检查协议
+    scheme = parsed.scheme.lower()
+    if scheme in _DANGEROUS_SCHEMES:
+        return False, f'Dangerous scheme: {scheme}'
+
+    if scheme not in ('http', 'https'):
+        return False, f'Unsupported scheme: {scheme}'
+
+    # 检查主机名
+    hostname = parsed.hostname
+    if not hostname:
+        return False, 'Missing hostname'
+
+    # 检查是否是危险主机名
+    dangerous_hosts = {'localhost', '0.0.0.0', '127.0.0.1', '::1'}
+    if hostname.lower() in dangerous_hosts:
+        return False, f'Dangerous hostname: {hostname}'
+
+    # 解析 IP 地址并检查是否在私有网络
+    try:
+        ip = ipaddress.ip_address(hostname)
+        for network in _PRIVATE_NETWORKS:
+            if ip in network:
+                return False, f'Private network address: {ip}'
+    except ValueError:
+        # 不是 IP 地址，尝试 DNS 解析检查
+        # 这里可以添加 DNS 解析检查，但为了避免复杂性，暂时跳过
+        pass
+
+    return True, ''
+
+
 @workflow_node('http_request')
 class HTTPRequestNode(WorkflowNode):
     """HTTP request node - make HTTP API calls"""
@@ -20,11 +82,30 @@ class HTTPRequestNode(WorkflowNode):
         import aiohttp
 
         url = self.get_config('url', '')
-        method = self.get_config('method', 'GET')
+        method = self.get_config('method', 'GET').upper()
         timeout = self.get_config('timeout', 30)
         content_type = self.get_config('content_type', 'application/json')
+        allow_redirects = self.get_config('allow_redirects', False)  # 默认禁用重定向
 
-        headers = inputs.get('headers', {})
+        # 限制超时时间
+        timeout = min(max(timeout, 1), 120)
+
+        if not url:
+            return {'response': None, 'status_code': 0, 'headers': {}, 'error': 'No URL provided'}
+
+        # 安全检查 URL
+        is_safe, error_msg = _is_safe_url(url)
+        if not is_safe:
+            logger.warning('Unsafe URL blocked: %s - %s', url, error_msg)
+            return {'response': None, 'status_code': 0, 'headers': {}, 'error': f'Unsafe URL: {error_msg}'}
+
+        # 验证 HTTP 方法
+        allowed_methods = {'GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS'}
+        if method not in allowed_methods:
+            return {'response': None, 'status_code': 0, 'headers': {}, 'error': f'Invalid method: {method}'}
+
+        # 创建 headers 副本，避免修改输入
+        headers = dict(inputs.get('headers', {}))
         headers['Content-Type'] = content_type
 
         auth_type = self.get_config('auth_type', 'none')
@@ -38,6 +119,8 @@ class HTTPRequestNode(WorkflowNode):
 
         body = inputs.get('body')
 
+        logger.info('HTTP %s %s (timeout=%s)', method, url, timeout)
+
         try:
             async with aiohttp.ClientSession() as session:
                 async with session.request(
@@ -47,16 +130,24 @@ class HTTPRequestNode(WorkflowNode):
                     data=body if content_type != 'application/json' else None,
                     headers=headers,
                     timeout=aiohttp.ClientTimeout(total=timeout),
+                    allow_redirects=allow_redirects,
                 ) as response:
                     try:
                         response_data = await response.json()
                     except Exception:
                         response_data = await response.text()
 
+                    logger.info('HTTP %s %s -> %d', method, url, response.status)
+
                     return {
                         'response': response_data,
                         'status_code': response.status,
                         'headers': dict(response.headers),
+                        'error': None,
                     }
+        except aiohttp.ClientError as e:
+            logger.error('HTTP request failed: %s', e)
+            return {'response': None, 'status_code': 0, 'headers': {}, 'error': f'HTTP error: {e}'}
         except Exception as e:
-            return {'response': None, 'status_code': 0, 'headers': {}, 'error': str(e)}
+            logger.error('HTTP request unexpected error: %s', e)
+            return {'response': None, 'status_code': 0, 'headers': {}, 'error': f'Unexpected error: {e}'}

src/langbot/pkg/workflow/nodes/langflow_flow.py

@@ -22,12 +22,15 @@ class LangflowFlowNode(WorkflowNode):
         flow_id = self.get_config('flow_id', '')
         input_value = inputs.get('input_value', '')
 
+        # Safe API key truncation
+        masked_key = f'{api_key[:4]}...{api_key[-4:]}' if len(api_key) > 8 else '***' if api_key else ''
+
         return {
             'result': None,
             'success': False,
             '_debug': {
                 'base_url': base_url,
-                'api_key': api_key[:8] + '...' if api_key else '',
+                'api_key': masked_key,
                 'flow_id': flow_id,
                 'input_value': input_value,
             },

src/langbot/pkg/workflow/nodes/llm_call.py

@@ -282,6 +282,13 @@ class LLMCallNode(WorkflowNode):
         # Remove CoT content (always remove to avoid leaking internal reasoning)
         response_text = self._remove_think_content(response_text)
 
+        # Initialize usage default
+        usage = {
+            'prompt_tokens': 0,
+            'completion_tokens': 0,
+            'total_tokens': 0,
+        }
+
         # Apply content safety filter
         response_text, is_blocked, filter_notice = self._apply_content_filter(response_text)
         if is_blocked:

src/langbot/pkg/workflow/nodes/message_trigger.py

@@ -32,12 +32,13 @@ class MessageTriggerNode(WorkflowNode):
                 'context': msg_ctx.model_dump(),
             }
 
+        # Use safe variable access with fallback
         return {
-            'message': context.get_variable('message', ''),
+            'message': context.get_variable('message') or '',
-            'sender_id': context.get_variable('sender_id', ''),
+            'sender_id': context.get_variable('sender_id') or '',
-            'sender_name': context.get_variable('sender_name', ''),
+            'sender_name': context.get_variable('sender_name') or '',
-            'platform': context.get_variable('platform', ''),
+            'platform': context.get_variable('platform') or '',
-            'conversation_id': context.get_variable('conversation_id', ''),
+            'conversation_id': context.get_variable('conversation_id') or '',
-            'is_group': context.get_variable('is_group', False),
+            'is_group': context.get_variable('is_group') or False,
-            'context': context.trigger_data,
+            'context': context.trigger_data or {},
         }

src/langbot/pkg/workflow/nodes/parameter_extractor.py

@@ -5,11 +5,15 @@ Node metadata is loaded from: ../../templates/metadata/nodes/parameter_extractor
 
 from __future__ import annotations
 
+import json
+import logging
 from typing import Any
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
 from ..node import WorkflowNode, workflow_node
 
+logger = logging.getLogger(__name__)
+
 @workflow_node('parameter_extractor')
 class ParameterExtractorNode(WorkflowNode):
     """Parameter extractor node - extract structured parameters from text"""
@@ -18,10 +22,112 @@ class ParameterExtractorNode(WorkflowNode):
     icon: str = 'Variable'
 
     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
+        # Get input text
+        input_text = inputs.get('input') or inputs.get('message') or inputs.get('content') or ''
+        input_text = str(input_text) if input_text is not None else ''
+
+        # Get configuration
         param_defs = self.get_config('parameters', [])
+        model_id = self.get_config('model', '')
+        system_prompt = self.get_config('system_prompt', '')
 
-        extracted = {}
+        if not input_text.strip():
+            return {
+                'parameters': {},
+                'extraction_success': False,
+                'error': 'Empty input',
+            }
+
+        if not param_defs:
+            return {
+                'parameters': {},
+                'extraction_success': False,
+                'error': 'No parameters configured',
+            }
+
+        # Build parameter schema for LLM prompt
+        param_schema = []
         for param in param_defs:
-            extracted[param.get('name', '')] = None
+            schema_item = {
+                'name': param.get('name', ''),
+                'type': param.get('type', 'string'),
+                'description': param.get('description', ''),
+                'required': param.get('required', False),
+            }
+            param_schema.append(schema_item)
 
-        return {'parameters': extracted, 'extraction_success': False}
+        # Build extraction prompt
+        if not system_prompt:
+            system_prompt = (
+                f'Extract the following parameters from the user\'s text as JSON. '
+                f'Respond with ONLY a valid JSON object containing the extracted parameters.\n\n'
+                f'Parameters to extract:\n'
+                f'{json.dumps(param_schema, indent=2, ensure_ascii=False)}\n\n'
+                f'Respond with a JSON object like: {{"param_name": "value", ...}}'
+            )
+
+        # Call LLM for extraction
+        if self.ap and model_id:
+            try:
+                # Get model (same as llm_call.py)
+                runtime_model = await self.ap.model_mgr.get_model_by_uuid(model_id)
+
+                # Build messages
+                from langbot_plugin.api.entities.builtin.provider.message import Message
+                messages = []
+                if system_prompt:
+                    messages.append(Message(role='system', content=system_prompt))
+                messages.append(Message(role='user', content=input_text))
+
+                # Invoke LLM (same as llm_call.py)
+                result_message = await runtime_model.provider.invoke_llm(
+                    query=None,
+                    model=runtime_model,
+                    messages=messages,
+                    funcs=None,
+                    extra_args={},
+                )
+
+                # Extract response text
+                response_text = ''
+                if isinstance(result_message.content, str):
+                    response_text = result_message.content
+                elif isinstance(result_message.content, list):
+                    for elem in result_message.content:
+                        if hasattr(elem, 'text') and elem.text:
+                            response_text += elem.text
+                        elif isinstance(elem, str):
+                            response_text += elem
+
+                response_text = response_text.strip()
+
+                # Parse JSON response
+                try:
+                    extracted = json.loads(response_text)
+                    return {
+                        'parameters': extracted,
+                        'extraction_success': True,
+                        'raw_response': response_text[:500],
+                    }
+                except json.JSONDecodeError as e:
+                    logger.error('ParameterExtractorNode JSON parse error: %s', e)
+                    return {
+                        'parameters': {},
+                        'extraction_success': False,
+                        'error': f'Failed to parse JSON: {e}',
+                        'raw_response': response_text[:500],
+                    }
+
+            except Exception as e:
+                logger.error('ParameterExtractorNode LLM error: %s', e, exc_info=True)
+                return {
+                    'parameters': {},
+                    'extraction_success': False,
+                    'error': f'LLM error: {e}',
+                }
+        else:
+            return {
+                'parameters': {},
+                'extraction_success': False,
+                'error': 'Missing model configuration',
+            }

src/langbot/pkg/workflow/nodes/question_classifier.py

@@ -5,11 +5,15 @@ Node metadata is loaded from: ../../templates/metadata/nodes/question_classifier
 
 from __future__ import annotations
 
+import json
+import logging
 from typing import Any
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
 from ..node import WorkflowNode, workflow_node
 
+logger = logging.getLogger(__name__)
+
 @workflow_node('question_classifier')
 class QuestionClassifierNode(WorkflowNode):
     """Question classifier node - classify user questions into categories"""
@@ -17,13 +21,110 @@ class QuestionClassifierNode(WorkflowNode):
     category = 'process'
 
     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        categories = self.get_config('categories', [])
+        # Get input text
+        input_text = inputs.get('input') or inputs.get('message') or inputs.get('content') or ''
+        input_text = str(input_text) if input_text is not None else ''
 
-        if categories:
+        # Get configuration
+        categories = self.get_config('categories', [])
+        model_id = self.get_config('model', '')
+        system_prompt = self.get_config('system_prompt', '')
+
+        if not input_text.strip():
             return {
-                'category': categories[0].get('name', 'unknown'),
+                'category': 'unknown',
-                'confidence': 0.8,
+                'confidence': 0.0,
-                'all_scores': {cat.get('name'): 0.1 for cat in categories},
+                'all_scores': {},
+                'error': 'Empty input',
             }
 
-        return {'category': 'unknown', 'confidence': 0.0, 'all_scores': {}}
+        if not categories:
+            return {
+                'category': 'unknown',
+                'confidence': 0.0,
+                'all_scores': {},
+                'error': 'No categories configured',
+            }
+
+        # Build category list for LLM prompt
+        category_names = [cat.get('name', '') for cat in categories if cat.get('name')]
+
+        # Build classification prompt
+        if not system_prompt:
+            system_prompt = (
+                f'You are a question classifier. Classify the user\'s question into one of these categories: '
+                f'{", ".join(category_names)}. '
+                f'Respond with ONLY the category name, nothing else.'
+            )
+
+        # Call LLM for classification
+        if self.ap and model_id:
+            try:
+                # Get model (same as llm_call.py)
+                runtime_model = await self.ap.model_mgr.get_model_by_uuid(model_id)
+
+                # Build messages
+                from langbot_plugin.api.entities.builtin.provider.message import Message
+                messages = []
+                if system_prompt:
+                    messages.append(Message(role='system', content=system_prompt))
+                messages.append(Message(role='user', content=input_text))
+
+                # Invoke LLM (same as llm_call.py)
+                result_message = await runtime_model.provider.invoke_llm(
+                    query=None,
+                    model=runtime_model,
+                    messages=messages,
+                    funcs=None,
+                    extra_args={},
+                )
+
+                # Extract response text
+                response_text = ''
+                if isinstance(result_message.content, str):
+                    response_text = result_message.content
+                elif isinstance(result_message.content, list):
+                    for elem in result_message.content:
+                        if hasattr(elem, 'text') and elem.text:
+                            response_text += elem.text
+                        elif isinstance(elem, str):
+                            response_text += elem
+
+                response_text = response_text.strip()
+
+                # Find matching category
+                matched_category = None
+                for cat in categories:
+                    if cat.get('name', '').lower() == response_text.lower():
+                        matched_category = cat
+                        break
+
+                if matched_category:
+                    return {
+                        'category': matched_category['name'],
+                        'confidence': 0.9,
+                        'all_scores': {cat.get('name', ''): 0.1 for cat in categories},
+                    }
+                else:
+                    # Default to first category if no match
+                    return {
+                        'category': category_names[0] if category_names else 'unknown',
+                        'confidence': 0.5,
+                        'all_scores': {cat.get('name', ''): 0.1 for cat in categories},
+                    }
+
+            except Exception as e:
+                logger.error('QuestionClassifierNode LLM error: %s', e, exc_info=True)
+                return {
+                    'category': category_names[0] if category_names else 'unknown',
+                    'confidence': 0.0,
+                    'all_scores': {},
+                    'error': f'LLM error: {e}',
+                }
+        else:
+            return {
+                'category': category_names[0] if category_names else 'unknown',
+                'confidence': 0.0,
+                'all_scores': {},
+                'error': 'Missing model configuration',
+            }

src/langbot/pkg/workflow/nodes/reply_message.py

@@ -38,9 +38,17 @@ class ReplyMessageNode(WorkflowNode):
         if template:
             message = template
             for key, value in inputs.items():
-                message = message.replace(f'{{{{{key}}}}}', str(value))
+                try:
+                    message = message.replace(f'{{{{{key}}}}}', str(value) if value is not None else '')
+                except Exception:
+                    pass
             for key, value in context.variables.items():
-                message = message.replace(f'{{{{variables.{key}}}}}', str(value))
+                try:
+                    message = message.replace(f'{{{{variables.{key}}}}}', str(value) if value is not None else '')
+                except Exception:
+                    pass
+
+        message_str = str(message) if message is not None else ''
 
         logger.info(
             'ReplyMessageNode resolved message',
@@ -48,13 +56,13 @@ class ReplyMessageNode(WorkflowNode):
                 'node_id': self.node_id,
                 'execution_id': context.execution_id,
                 'input_keys': list(inputs.keys()),
-                'message_preview': str(message)[:200],
+                'message_preview': message_str[:200],
                 'has_template': bool(template),
                 'session_id': context.session_id,
             },
         )
 
-        if not str(message).strip():
+        if not message_str.strip():
             logger.warning(
                 'ReplyMessageNode has empty message after resolution',
                 extra={
@@ -65,16 +73,28 @@ class ReplyMessageNode(WorkflowNode):
             )
 
         # 实际发送消息
+        send_success = False
+        send_error = None
         if self.ap:
-            from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+            try:
+                from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
 
-            message_chain = MessageChain([Plain(text=str(message))])
+                message_chain = MessageChain([Plain(text=message_str)])
-            await self.ap.platform_mgr.websocket_proxy_bot.adapter.send_message(
+                target_type = getattr(context, 'target_type', 'person') or 'person'
-                target_type='person',
+                session_id = context.session_id or 'unknown'
-                target_id=f'websocket_{context.session_id}',
+                target_id = f'websocket_{session_id}'
-                message=message_chain,
+
-            )
+                await self.ap.platform_mgr.websocket_proxy_bot.adapter.send_message(
+                    target_type=target_type,
+                    target_id=target_id,
+                    message=message_chain,
+                )
+                send_success = True
+            except Exception as e:
+                send_error = str(e)
+                logger.error('ReplyMessageNode send message failed: %s', e, exc_info=True)
         else:
+            send_error = 'Missing application instance'
             logger.warning(
                 'ReplyMessageNode missing application instance',
                 extra={
@@ -83,4 +103,8 @@ class ReplyMessageNode(WorkflowNode):
                 },
             )
 
-        return {'status': 'sent', 'message_id': f'reply_{context.execution_id}'}
+        return {
+            'status': 'sent' if send_success else 'failed',
+            'message_preview': message_str[:200],
+            'error': send_error,
+        }

src/langbot/pkg/workflow/nodes/send_message.py

@@ -5,11 +5,14 @@ Node metadata is loaded from: ../../templates/metadata/nodes/send_message.yaml
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
 from ..node import WorkflowNode, workflow_node
 
+logger = logging.getLogger(__name__)
+
 @workflow_node('send_message')
 class SendMessageNode(WorkflowNode):
     """Send message node - send message to a target"""
@@ -17,4 +20,52 @@ class SendMessageNode(WorkflowNode):
     category = 'action'
 
     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        return {'status': 'sent', 'message_id': f'msg_{context.execution_id}'}
+        # Get message content from inputs
+        message = inputs.get('message') or inputs.get('content') or inputs.get('input') or ''
+        message = str(message) if message is not None else ''
+
+        # Get target configuration
+        target_type = self.get_config('target_type', 'person')
+        target_id = self.get_config('target_id', '')
+        
+        # If no target_id configured, use session_id from context
+        if not target_id:
+            target_id = f'{context.session_id or "unknown"}'
+
+        if not message.strip():
+            logger.warning('SendMessageNode has empty message')
+            return {
+                'status': 'failed',
+                'error': 'Empty message',
+                'message_preview': '',
+            }
+
+        # Send message if application instance is available
+        send_success = False
+        send_error = None
+        if self.ap:
+            try:
+                from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+                message_chain = MessageChain([Plain(text=message)])
+                await self.ap.platform_mgr.websocket_proxy_bot.adapter.send_message(
+                    target_type=target_type,
+                    target_id=target_id,
+                    message=message_chain,
+                )
+                send_success = True
+                logger.info('SendMessageNode sent message to %s:%s', target_type, target_id)
+            except Exception as e:
+                send_error = str(e)
+                logger.error('SendMessageNode send failed: %s', e, exc_info=True)
+        else:
+            send_error = 'Missing application instance'
+            logger.warning('SendMessageNode missing application instance')
+
+        return {
+            'status': 'sent' if send_success else 'failed',
+            'message_preview': message[:200],
+            'target_type': target_type,
+            'target_id': target_id,
+            'error': send_error,
+        }

src/langbot/pkg/workflow/nodes/wait.py

@@ -5,11 +5,18 @@ Node metadata is loaded from: ../../templates/metadata/nodes/wait.yaml
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
 from langbot_plugin.api.entities.builtin.workflow import ExecutionContext
 from ..node import WorkflowNode, workflow_node
 
+logger = logging.getLogger(__name__)
+
+# 最大等待时间（秒）
+_MAX_WAIT_SECONDS = 300  # 5 分钟
+
+
 @workflow_node('wait')
 class WaitNode(WorkflowNode):
     """Wait node - pause execution for a duration"""
@@ -22,11 +29,22 @@ class WaitNode(WorkflowNode):
         duration = self.get_config('duration', 1)
         duration_type = self.get_config('duration_type', 'seconds')
 
+        # 转换为秒
         if duration_type == 'minutes':
             duration *= 60
         elif duration_type == 'hours':
             duration *= 3600
 
+        # 限制最大等待时间
+        if duration > _MAX_WAIT_SECONDS:
+            logger.warning('Wait duration %s exceeds maximum %s, capping to %s',
+                          duration, _MAX_WAIT_SECONDS, _MAX_WAIT_SECONDS)
+            duration = _MAX_WAIT_SECONDS
+
+        # 确保 duration 为正数
+        duration = max(0, duration)
+
+        logger.info('Waiting for %.2f seconds', duration)
         await asyncio.sleep(duration)
 
-        return {'output': inputs.get('input')}
+        return {'output': inputs.get('input'), 'waited_seconds': duration}

src/langbot/pkg/workflow/nodes/webhook_trigger.py

@@ -17,11 +17,17 @@ class WebhookTriggerNode(WorkflowNode):
     category = 'trigger'
 
     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        trigger_data = context.trigger_data
+        # Safe access to trigger_data which may be None
+        trigger_data = context.trigger_data or {}
+
+        # Filter sensitive headers (Authorization, Cookie, etc.)
+        headers = trigger_data.get('headers', {})
+        safe_headers = {k: v for k, v in headers.items()
+                       if k.lower() not in ('authorization', 'cookie', 'x-api-key', 'x-secret')}
 
         return {
             'body': trigger_data.get('body', {}),
-            'headers': trigger_data.get('headers', {}),
+            'headers': safe_headers,
             'query': trigger_data.get('query', {}),
             'method': trigger_data.get('method', 'POST'),
         }