n8n.py

"""
title: n8n Pipeline with StreamingResponse Support
author: owndev
author_url: https://github.com/owndev/
project_url: https://github.com/owndev/Open-WebUI-Functions
funding_url: https://github.com/sponsors/owndev
n8n_template: https://github.com/owndev/Open-WebUI-Functions/blob/main/pipelines/n8n/Open_WebUI_Test_Agent_Streaming.json
version: 2.3.0
required_open_webui_version: 0.8.0
license: Apache License 2.0
description: An optimized streaming-enabled pipeline for interacting with N8N workflows, consistent response handling for both streaming and non-streaming modes, robust error handling, and simplified status management. Supports Server-Sent Events (SSE) streaming and various N8N workflow formats. Now includes configurable AI Agent tool usage display with three verbosity levels (minimal, compact, detailed) and customizable length limits for tool inputs/outputs (non-streaming mode only).
features:
  - Integrates with N8N for seamless streaming communication.
  - Uses FastAPI StreamingResponse for real-time streaming.
  - Enables real-time interaction with N8N workflows.
  - Provides configurable status emissions and chunk streaming.
  - Cloudflare Access support for secure communication.
  - Encrypted storage of sensitive API keys.
  - Fallback support for non-streaming responses.
  - Compatible with Open WebUI streaming architecture.
  - Displays N8N AI Agent tool usage with configurable verbosity (non-streaming mode only).
  - Three display modes: minimal (tool names only), compact (names + preview), detailed (full collapsible sections).
  - Customizable length limits for tool inputs and outputs.
  - Shows tool calls, inputs, and results from intermediateSteps in non-streaming mode (N8N limitation - streaming responses do not include intermediateSteps).
"""

from typing import (
    Optional,
    Callable,
    Awaitable,
    Any,
    Dict,
    AsyncIterator,
    Union,
    Generator,
    Iterator,
)
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field, GetCoreSchemaHandler
from starlette.background import BackgroundTask
from cryptography.fernet import Fernet, InvalidToken
import aiohttp
import os
import base64
import hashlib
import logging
import json
import asyncio
from open_webui.env import AIOHTTP_CLIENT_TIMEOUT, SRC_LOG_LEVELS
from pydantic_core import core_schema
import time
import re


# Simplified encryption implementation with automatic handling
class EncryptedStr(str):
    """A string type that automatically handles encryption/decryption"""

    @classmethod
    def _get_encryption_key(cls) -> Optional[bytes]:
        """
        Generate encryption key from WEBUI_SECRET_KEY if available
        Returns None if no key is configured
        """
        secret = os.getenv("WEBUI_SECRET_KEY")
        if not secret:
            return None

        hashed_key = hashlib.sha256(secret.encode()).digest()
        return base64.urlsafe_b64encode(hashed_key)

    @classmethod
    def encrypt(cls, value: str) -> str:
        """
        Encrypt a string value if a key is available
        Returns the original value if no key is available
        """
        if not value or value.startswith("encrypted:"):
            return value

        key = cls._get_encryption_key()
        if not key:  # No encryption if no key
            return value

        f = Fernet(key)
        encrypted = f.encrypt(value.encode())
        return f"encrypted:{encrypted.decode()}"

    @classmethod
    def decrypt(cls, value: str) -> str:
        """
        Decrypt an encrypted string value if a key is available
        Returns the original value if no key is available or decryption fails
        """
        if not value or not value.startswith("encrypted:"):
            return value

        key = cls._get_encryption_key()
        if not key:  # No decryption if no key
            return value[len("encrypted:") :]  # Return without prefix

        try:
            encrypted_part = value[len("encrypted:") :]
            f = Fernet(key)
            decrypted = f.decrypt(encrypted_part.encode())
            return decrypted.decode()
        except (InvalidToken, Exception):
            return value

    # Pydantic integration
    @classmethod
    def __get_pydantic_core_schema__(
        cls, _source_type: Any, _handler: GetCoreSchemaHandler
    ) -> core_schema.CoreSchema:
        return core_schema.union_schema(
            [
                core_schema.is_instance_schema(cls),
                core_schema.chain_schema(
                    [
                        core_schema.str_schema(),
                        core_schema.no_info_plain_validator_function(
                            lambda value: cls(cls.encrypt(value) if value else value)
                        ),
                    ]
                ),
            ],
            serialization=core_schema.plain_serializer_function_ser_schema(
                lambda instance: str(instance)
            ),
        )


# Helper functions for resource cleanup
async def cleanup_response(
    response: Optional[aiohttp.ClientResponse],
    session: Optional[aiohttp.ClientSession],
) -> None:
    """
    Clean up the response and session objects.

    Args:
        response: The ClientResponse object to close
        session: The ClientSession object to close
    """
    if response:
        response.close()
    if session:
        await session.close()


async def stream_processor(
    content: aiohttp.StreamReader,
    __event_emitter__=None,
    response: Optional[aiohttp.ClientResponse] = None,
    session: Optional[aiohttp.ClientSession] = None,
    logger: Optional[logging.Logger] = None,
) -> AsyncIterator[str]:
    """
    Process streaming content from n8n and yield chunks for StreamingResponse.

    Args:
        content: The streaming content from the response
        __event_emitter__: Optional event emitter for status updates
        response: The response object for cleanup
        session: The session object for cleanup
        logger: Logger for debugging

    Yields:
        String content from the streaming response
    """
    try:
        if logger:
            logger.info("Starting stream processing...")

        buffer = ""
        # Attempt to read preserve flag later via closure if needed
        async for chunk_bytes in content:
            chunk_str = chunk_bytes.decode("utf-8", errors="ignore")
            if not chunk_str:
                continue
            buffer += chunk_str

            # Process complete lines (retain trailing newline info)
            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                had_newline = True
                original_line = line  # without \n
                if line.endswith("\r"):
                    line = line[:-1]

                if logger:
                    logger.debug(f"Raw line received: {repr(line)}")

                # Preserve blank lines
                if line == "":
                    yield "\n"
                    continue

                content_text = ""

                if line.startswith("data: "):
                    data_part = line[6:]
                    if logger:
                        logger.debug(f"SSE data part: {repr(data_part)}")
                    if data_part == "[DONE]":
                        if logger:
                            logger.debug("Received [DONE] signal")
                        buffer = ""
                        break
                    try:
                        event_data = json.loads(data_part)
                        if logger:
                            logger.debug(f"Parsed SSE JSON: {event_data}")
                        for key in ("content", "text", "output", "data"):
                            val = event_data.get(key)
                            if isinstance(val, str) and val:
                                content_text = val
                                break
                    except json.JSONDecodeError:
                        content_text = data_part
                        if logger:
                            logger.debug(
                                f"Using raw data as content: {repr(content_text)}"
                            )
                elif not line.startswith(":"):
                    # Plain text (non-SSE)
                    content_text = original_line
                    if logger:
                        logger.debug(f"Plain text content: {repr(content_text)}")

                if content_text:
                    if not content_text.endswith("\n"):
                        content_text += "\n"
                    if logger:
                        logger.debug(f"Yielding content: {repr(content_text)}")
                    yield content_text

        # Send completion status update when streaming is done
        if __event_emitter__:
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "status": "complete",
                        "description": "N8N streaming completed successfully",
                        "done": True,
                    },
                }
            )

        if logger:
            logger.info("Stream processing completed successfully")

    except Exception as e:
        if logger:
            logger.error(f"Error processing stream: {e}")

        # Send error status update
        if __event_emitter__:
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "status": "error",
                        "description": f"N8N streaming error: {str(e)}",
                        "done": True,
                    },
                }
            )
        raise
    finally:
        # Always attempt to close response and session to avoid resource leaks
        await cleanup_response(response, session)


class Pipe:
    class Valves(BaseModel):
        N8N_URL: str = Field(
            default="https://<your-endpoint>/webhook/<your-webhook>",
            description="URL for the N8N webhook",
        )
        N8N_BEARER_TOKEN: EncryptedStr = Field(
            default="",
            description="Bearer token for authenticating with the N8N webhook",
            json_schema_extra={"input": {"type": "password"}},
        )
        INPUT_FIELD: str = Field(
            default="chatInput",
            description="Field name for the input message in the N8N payload",
        )
        RESPONSE_FIELD: str = Field(
            default="output",
            description="Field name for the response message in the N8N payload",
        )
        SEND_CONVERSATION_HISTORY: bool = Field(
            default=False,
            description="Whether to include conversation history when sending requests to N8N",
        )
        TOOL_DISPLAY_VERBOSITY: str = Field(
            default="detailed",
            description="Verbosity level for tool usage display: 'minimal' (only tool names), 'compact' (names + short preview), 'detailed' (full info with collapsible sections)",
        )
        TOOL_INPUT_MAX_LENGTH: int = Field(
            default=500,
            description="Maximum length for tool input display (0 = unlimited). Longer inputs will be truncated.",
        )
        TOOL_OUTPUT_MAX_LENGTH: int = Field(
            default=500,
            description="Maximum length for tool output/observation display (0 = unlimited). Longer outputs will be truncated.",
        )
        CF_ACCESS_CLIENT_ID: EncryptedStr = Field(
            default="",
            description="Only if behind Cloudflare: https://developers.cloudflare.com/cloudflare-one/identity/service-tokens/",
            json_schema_extra={"input": {"type": "password"}},
        )
        CF_ACCESS_CLIENT_SECRET: EncryptedStr = Field(
            default="",
            description="Only if behind Cloudflare: https://developers.cloudflare.com/cloudflare-one/identity/service-tokens/",
            json_schema_extra={"input": {"type": "password"}},
        )

    def __init__(self):
        self.name = "N8N Agent"
        self.valves = self.Valves()
        self.log = logging.getLogger("n8n_streaming_pipeline")
        self.log.setLevel(SRC_LOG_LEVELS.get("OPENAI", logging.INFO))

    def _format_tool_calls_section(
        self, intermediate_steps: list, for_streaming: bool = False
    ) -> str:
        """
        Creates a formatted tool calls section using collapsible details elements.

        Args:
            intermediate_steps: List of intermediate step objects from N8N response
            for_streaming: If True, format for streaming (with escaping), else for regular response

        Returns:
            Formatted tool calls section with HTML details elements
        """
        if not intermediate_steps:
            return ""

        verbosity = self.valves.TOOL_DISPLAY_VERBOSITY.lower()
        input_max_len = self.valves.TOOL_INPUT_MAX_LENGTH
        output_max_len = self.valves.TOOL_OUTPUT_MAX_LENGTH

        # Helper function to truncate text
        def truncate_text(text: str, max_length: int) -> str:
            if max_length <= 0 or len(text) <= max_length:
                return text
            return text[:max_length] + "..."

        # Minimal mode: just list tool names
        if verbosity == "minimal":
            tool_names = []
            for i, step in enumerate(intermediate_steps, 1):
                if isinstance(step, dict):
                    tool_name = step.get("action", {}).get("tool", "Unknown Tool")
                    tool_names.append(f"{i}. {tool_name}")

            tool_list = "\\n" if for_streaming else "\n"
            tool_list = tool_list.join(tool_names)

            if for_streaming:
                return f"\\n\\n<details>\\n<summary>🛠️ Tool Calls ({len(intermediate_steps)} steps)</summary>\\n\\n{tool_list}\\n\\n</details>\\n"
            else:
                return f"\n\n<details>\n<summary>🛠️ Tool Calls ({len(intermediate_steps)} steps)</summary>\n\n{tool_list}\n\n</details>\n"

        # Compact mode: tool names with short preview
        if verbosity == "compact":
            tool_summaries = []
            for i, step in enumerate(intermediate_steps, 1):
                if not isinstance(step, dict):
                    continue

                action = step.get("action", {})
                observation = step.get("observation", "")
                tool_name = action.get("tool", "Unknown Tool")

                # Get short preview of output
                preview = ""
                if observation:
                    obs_str = str(observation)
                    # If output_max_len is 0 (unlimited), use a reasonable default preview length for compact mode
                    # Otherwise, use the configured limit
                    if output_max_len > 0:
                        preview_len = min(100, output_max_len)
                    else:
                        preview_len = 100  # Default preview length for compact mode when unlimited
                    preview = truncate_text(obs_str, preview_len)

                summary = f"**{i}. {tool_name}**"
                if preview:
                    summary += f" → {preview}"
                tool_summaries.append(summary)

            summary_text = "\\n" if for_streaming else "\n"
            summary_text = summary_text.join(tool_summaries)

            if for_streaming:
                return f"\\n\\n<details>\\n<summary>🛠️ Tool Calls ({len(intermediate_steps)} steps)</summary>\\n\\n{summary_text}\\n\\n</details>\\n"
            else:
                return f"\n\n<details>\n<summary>🛠️ Tool Calls ({len(intermediate_steps)} steps)</summary>\n\n{summary_text}\n\n</details>\n"

        # Detailed mode: full collapsible sections (default)
        tool_entries = []

        for i, step in enumerate(intermediate_steps, 1):
            if not isinstance(step, dict):
                continue

            action = step.get("action", {})
            observation = step.get("observation", "")

            tool_name = action.get("tool", "Unknown Tool")
            tool_input = action.get("toolInput", {})
            tool_call_id = action.get("toolCallId", "")
            log_message = action.get("log", "")

            # Build individual tool call details
            tool_info = []
            tool_info.append(f"🔧 **Tool:** {tool_name}")

            if tool_call_id:
                tool_info.append(f"🆔 **Call ID:** `{tool_call_id}`")

            # Format tool input
            if tool_input:
                try:
                    if isinstance(tool_input, dict):
                        input_json = json.dumps(tool_input, indent=2)

                        # Apply max length limit
                        if input_max_len > 0:
                            input_json = truncate_text(input_json, input_max_len)

                        if for_streaming:
                            # Escape for streaming
                            input_json = (
                                input_json.replace("\\", "\\\\")
                                .replace('"', '\\"')
                                .replace("\n", "\\n")
                            )
                            tool_info.append(
                                f"📥 **Input:**\\n```json\\n{input_json}\\n```"
                            )
                        else:
                            tool_info.append(
                                f"📥 **Input:**\n```json\n{input_json}\n```"
                            )
                    else:
                        input_str = str(tool_input)
                        if input_max_len > 0:
                            input_str = truncate_text(input_str, input_max_len)
                        tool_info.append(f"📥 **Input:** `{input_str}`")
                except Exception:
                    input_str = str(tool_input)
                    if input_max_len > 0:
                        input_str = truncate_text(input_str, input_max_len)
                    tool_info.append(f"📥 **Input:** `{input_str}`")

            # Format observation/result
            if observation:
                try:
                    # Try to parse as JSON for better formatting
                    if isinstance(observation, str) and (
                        observation.startswith("[") or observation.startswith("{")
                    ):
                        obs_json = json.loads(observation)
                        obs_formatted = json.dumps(obs_json, indent=2)

                        # Apply max length limit
                        if output_max_len > 0:
                            obs_formatted = truncate_text(obs_formatted, output_max_len)

                        if for_streaming:
                            obs_formatted = (
                                obs_formatted.replace("\\", "\\\\")
                                .replace('"', '\\"')
                                .replace("\n", "\\n")
                            )
                            tool_info.append(
                                f"📤 **Result:**\\n```json\\n{obs_formatted}\\n```"
                            )
                        else:
                            tool_info.append(
                                f"📤 **Result:**\n```json\n{obs_formatted}\n```"
                            )
                    else:
                        # Plain text observation
                        obs_str = str(observation)
                        # Apply configured limit (0 = unlimited, don't truncate)
                        obs_preview = (
                            truncate_text(obs_str, output_max_len)
                            if output_max_len > 0
                            else obs_str
                        )

                        if for_streaming:
                            obs_preview = (
                                obs_preview.replace("\\", "\\\\")
                                .replace('"', '\\"')
                                .replace("\n", "\\n")
                            )
                        tool_info.append(f"📤 **Result:** {obs_preview}")
                except Exception:
                    obs_str = str(observation)
                    # Apply configured limit (0 = unlimited, don't truncate)
                    obs_preview = (
                        truncate_text(obs_str, output_max_len)
                        if output_max_len > 0
                        else obs_str
                    )
                    tool_info.append(f"📤 **Result:** {obs_preview}")

            # Add log if available
            if log_message:
                log_preview = truncate_text(log_message, 200)
                tool_info.append(f"📝 **Log:** {log_preview}")

            # Create collapsible details for individual tool call
            tool_info_text = "\\n" if for_streaming else "\n"
            tool_info_text = tool_info_text.join(tool_info)

            if for_streaming:
                tool_entry = f"<details>\\n<summary>Step {i}: {tool_name}</summary>\\n\\n{tool_info_text}\\n\\n</details>"
            else:
                tool_entry = f"<details>\n<summary>Step {i}: {tool_name}</summary>\n\n{tool_info_text}\n\n</details>"

            tool_entries.append(tool_entry)

        # Combine all tool calls into main collapsible section
        if for_streaming:
            all_tools = "\\n\\n".join(tool_entries)
            result = f"\\n\\n<details>\\n<summary>🛠️ Tool Calls ({len(tool_entries)} steps)</summary>\\n\\n{all_tools}\\n\\n</details>\\n"
        else:
            all_tools = "\n\n".join(tool_entries)
            result = f"\n\n<details>\n<summary>🛠️ Tool Calls ({len(tool_entries)} steps)</summary>\n\n{all_tools}\n\n</details>\n"

        return result

    async def emit_simple_status(
        self,
        __event_emitter__: Callable[[dict], Awaitable[None]],
        status: str,
        message: str,
        done: bool = False,
    ):
        """Simplified status emission without intervals"""
        if __event_emitter__:
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "status": status,
                        "description": message,
                        "done": done,
                    },
                }
            )

    def extract_event_info(self, event_emitter):
        if not event_emitter or not event_emitter.__closure__:
            return None, None
        for cell in event_emitter.__closure__:
            if isinstance(request_info := cell.cell_contents, dict):
                chat_id = request_info.get("chat_id")
                message_id = request_info.get("message_id")
                return chat_id, message_id
        return None, None

    def get_headers(self) -> Dict[str, str]:
        """
        Constructs the headers for the API request.

        Returns:
            Dictionary containing the required headers for the API request.
        """
        headers = {"Content-Type": "application/json"}

        # Add bearer token if available
        bearer_token = EncryptedStr.decrypt(self.valves.N8N_BEARER_TOKEN)
        if bearer_token:
            headers["Authorization"] = f"Bearer {bearer_token}"

        # Add Cloudflare Access headers if available
        cf_client_id = EncryptedStr.decrypt(self.valves.CF_ACCESS_CLIENT_ID)
        if cf_client_id:
            headers["CF-Access-Client-Id"] = cf_client_id

        cf_client_secret = EncryptedStr.decrypt(self.valves.CF_ACCESS_CLIENT_SECRET)
        if cf_client_secret:
            headers["CF-Access-Client-Secret"] = cf_client_secret

        return headers

    def parse_n8n_streaming_chunk(self, chunk_text: str) -> Optional[str]:
        """Parse N8N streaming chunk and extract content, filtering out metadata"""
        if not chunk_text.strip():
            return None

        try:
            data = json.loads(chunk_text.strip())

            if isinstance(data, dict):
                # Check if this chunk contains intermediateSteps (will be handled separately)
                # Note: Don't skip chunks just because they have a type field
                chunk_type = data.get("type", "")

                # Skip only true metadata chunks that have no content or intermediateSteps
                if (
                    chunk_type in ["begin", "end", "error", "metadata"]
                    and "intermediateSteps" not in data
                ):
                    self.log.debug(f"Skipping N8N metadata chunk: {chunk_type}")
                    return None

                # Skip metadata-only chunks (but allow intermediateSteps)
                if (
                    "metadata" in data
                    and len(data) <= 2
                    and "intermediateSteps" not in data
                ):
                    return None

                # Extract content from various possible field names
                content = (
                    data.get("text")
                    or data.get("content")
                    or data.get("output")
                    or data.get("message")
                    or data.get("delta")
                    or data.get("data")
                    or data.get("response")
                    or data.get("result")
                )

                # Handle OpenAI-style streaming format
                if not content and "choices" in data:
                    choices = data.get("choices", [])
                    if choices and isinstance(choices[0], dict):
                        delta = choices[0].get("delta", {})
                        content = delta.get("content", "")

                if content:
                    self.log.debug(
                        f"Extracted content from JSON: {repr(content[:100])}"
                    )
                    return str(content)

                # Return non-metadata objects as strings (be more permissive)
                if not any(
                    key in data
                    for key in [
                        "type",
                        "metadata",
                        "nodeId",
                        "nodeName",
                        "timestamp",
                        "id",
                    ]
                ):
                    # For smaller models, return the entire object if it's simple
                    self.log.debug(
                        f"Returning entire object as content: {repr(str(data)[:100])}"
                    )
                    return str(data)

        except json.JSONDecodeError:
            # Handle plain text content - be more permissive
            stripped = chunk_text.strip()
            if stripped and not stripped.startswith("{"):
                self.log.debug(f"Returning plain text content: {repr(stripped[:100])}")
                return stripped

        return None

    def extract_content_from_mixed_stream(self, raw_text: str) -> str:
        """Extract content from mixed stream containing both metadata and content"""
        content_parts = []

        # First try to handle concatenated JSON objects
        if "{" in raw_text and "}" in raw_text:
            parts = raw_text.split("}{")

            for i, part in enumerate(parts):
                # Reconstruct valid JSON
                if i > 0:
                    part = "{" + part
                if i < len(parts) - 1:
                    part = part + "}"

                extracted = self.parse_n8n_streaming_chunk(part)
                if extracted:
                    content_parts.append(extracted)

        # If no JSON content found, treat as plain text
        if not content_parts:
            # Remove common streaming artifacts but preserve actual content
            cleaned = raw_text.strip()
            if (
                cleaned
                and not cleaned.startswith("data:")
                and not cleaned.startswith(":")
            ):
                self.log.debug(f"Using raw text as content: {repr(cleaned[:100])}")
                return cleaned

        return "".join(content_parts)

    def dedupe_system_prompt(self, text: str) -> str:
        """Remove duplicated content from the system prompt.

        Strategies:
        1. Detect full duplication where the prompt text is repeated twice consecutively.
        2. Remove duplicate lines (keeping first occurrence, preserving order & spacing where possible).
        3. Preserve blank lines but collapse consecutive duplicate non-blank lines.
        """
        if not text:
            return text

        original = text
        stripped = text.strip()

        # 1. Full duplication detection (exact repeat of first half == second half)
        half = len(stripped) // 2
        if len(stripped) % 2 == 0:
            first_half = stripped[:half].strip()
            second_half = stripped[half:].strip()
            if first_half and first_half == second_half:
                text = first_half

        # 2. Line-level dedupe
        lines = text.splitlines()
        seen = set()
        deduped = []
        for line in lines:
            key = line.strip()
            # Allow empty lines to pass through (formatting), but avoid repeating identical non-empty lines
            if key and key in seen:
                continue
            if key:
                seen.add(key)
            deduped.append(line)

        deduped_text = "\n".join(deduped).strip()

        if deduped_text != original.strip():
            self.log.debug("System prompt deduplicated")
        return deduped_text

    async def pipe(
        self,
        body: dict,
        __user__: Optional[dict] = None,
        __event_emitter__: Callable[[dict], Awaitable[None]] = None,
        __event_call__: Callable[[dict], Awaitable[dict]] = None,
    ) -> Union[str, Generator, Iterator, Dict[str, Any], StreamingResponse]:
        """
        Main method for sending requests to the N8N endpoint.

        Args:
            body: The request body containing messages and other parameters
            __event_emitter__: Optional event emitter function for status updates

        Returns:
            Response from N8N API, which could be a string, dictionary or streaming response
        """
        self.log.setLevel(SRC_LOG_LEVELS.get("OPENAI", logging.INFO))

        await self.emit_simple_status(
            __event_emitter__, "in_progress", f"Calling {self.name} ...", False
        )

        session = None
        n8n_response = ""
        messages = body.get("messages", [])

        # Verify a message is available
        if messages:
            question = messages[-1]["content"]
            if "Prompt: " in question:
                question = question.split("Prompt: ")[-1]
            try:
                # Extract chat_id and message_id
                chat_id, message_id = self.extract_event_info(__event_emitter__)

                self.log.info(f"Starting N8N workflow request for chat ID: {chat_id}")

                # Extract system prompt correctly
                system_prompt = ""
                if messages and messages[0].get("role") == "system":
                    system_prompt = self.dedupe_system_prompt(messages[0]["content"])

                # Optionally include full conversation history (controlled by valve)
                conversation_history = []
                if self.valves.SEND_CONVERSATION_HISTORY:
                    for msg in messages:
                        if msg.get("role") in ["user", "assistant"]:
                            conversation_history.append(
                                {"role": msg["role"], "content": msg["content"]}
                            )

                # Prepare payload for N8N workflow (improved version)
                payload = {
                    "systemPrompt": system_prompt,
                    # Include messages only when enabled in valves for privacy/control
                    "messages": (
                        conversation_history
                        if self.valves.SEND_CONVERSATION_HISTORY
                        else []
                    ),
                    "currentMessage": question,  # Current user message
                    "user_id": __user__.get("id") if __user__ else None,
                    "user_email": __user__.get("email") if __user__ else None,
                    "user_name": __user__.get("name") if __user__ else None,
                    "user_role": __user__.get("role") if __user__ else None,
                    "chat_id": chat_id,
                    "message_id": message_id,
                }
                # Keep backward compatibility
                payload[self.valves.INPUT_FIELD] = question

                # Get headers for the request
                headers = self.get_headers()

                # Create session with no timeout like in stream-example.py
                session = aiohttp.ClientSession(
                    trust_env=True,
                    timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT),
                )

                self.log.debug(f"Sending request to N8N: {self.valves.N8N_URL}")

                # Send status update via event emitter if available
                if __event_emitter__:
                    await __event_emitter__(
                        {
                            "type": "status",
                            "data": {
                                "status": "in_progress",
                                "description": "Sending request to N8N...",
                                "done": False,
                            },
                        }
                    )

                # Make the request
                request = session.post(
                    self.valves.N8N_URL, json=payload, headers=headers
                )

                response = await request.__aenter__()
                self.log.debug(f"Response status: {response.status}")
                self.log.debug(f"Response headers: {dict(response.headers)}")

                if response.status == 200:
                    # Enhanced streaming detection (n8n controls streaming)
                    content_type = response.headers.get("Content-Type", "").lower()

                    # Check for explicit streaming indicators
                    # Note: Don't rely solely on Transfer-Encoding: chunked as regular JSON can also be chunked
                    is_streaming = (
                        "text/event-stream" in content_type
                        or "application/x-ndjson" in content_type
                        or (
                            "application/json" in content_type
                            and response.headers.get("Transfer-Encoding") == "chunked"
                            and "Cache-Control" in response.headers
                            and "no-cache"
                            in response.headers.get("Cache-Control", "").lower()
                        )
                    )

                    # Additional check: if content-type is text/html or application/json without streaming headers, it's likely not streaming
                    if "text/html" in content_type:
                        is_streaming = False
                    elif (
                        "application/json" in content_type
                        and "Cache-Control" not in response.headers
                    ):
                        is_streaming = False

                    if is_streaming:
                        # Enhanced streaming like in stream-example.py
                        self.log.info("Processing streaming response from N8N")
                        n8n_response = ""
                        buffer = ""
                        completed_thoughts: list[str] = []
                        intermediate_steps = []  # Collect tool calls

                        try:
                            async for chunk in response.content.iter_any():
                                if not chunk:
                                    continue

                                text = chunk.decode(errors="ignore")
                                buffer += text

                                # Handle different streaming formats
                                if "{" in buffer and "}" in buffer:
                                    # Process complete JSON objects like in stream-example.py
                                    while True:
                                        start_idx = buffer.find("{")
                                        if start_idx == -1:
                                            break

                                        # Find matching closing brace
                                        brace_count = 0
                                        end_idx = -1

                                        for i in range(start_idx, len(buffer)):
                                            if buffer[i] == "{":
                                                brace_count += 1
                                            elif buffer[i] == "}":
                                                brace_count -= 1
                                                if brace_count == 0:
                                                    end_idx = i
                                                    break

                                        if end_idx == -1:
                                            # Incomplete JSON, wait for more data
                                            break

                                        # Extract and process the JSON chunk
                                        json_chunk = buffer[start_idx : end_idx + 1]
                                        buffer = buffer[end_idx + 1 :]

                                        # Try to parse the chunk as JSON to extract intermediateSteps
                                        # This must happen BEFORE parse_n8n_streaming_chunk filters out metadata
                                        # Future-proof: If N8N adds intermediateSteps support in streaming, this will work automatically
                                        try:
                                            parsed_chunk = json.loads(json_chunk)
                                            if isinstance(parsed_chunk, dict):
                                                # Extract intermediateSteps if present (future-proof for when N8N supports this)
                                                chunk_steps = parsed_chunk.get(
                                                    "intermediateSteps", []
                                                )
                                                if chunk_steps:
                                                    intermediate_steps.extend(
                                                        chunk_steps
                                                    )
                                                    self.log.info(
                                                        f"✓ Found {len(chunk_steps)} intermediate steps in streaming chunk"
                                                    )
                                        except json.JSONDecodeError:
                                            pass  # Continue with content parsing

                                        # Parse N8N streaming chunk for content
                                        content = self.parse_n8n_streaming_chunk(
                                            json_chunk
                                        )
                                        if content:
                                            # Normalize escaped newlines to actual newlines (like non-streaming)
                                            content = content.replace("\\n", "\n")

                                            # Just accumulate content without processing think blocks yet
                                            n8n_response += content

                                            # Emit delta without think block processing
                                            if __event_emitter__:
                                                await __event_emitter__(
                                                    {
                                                        "type": "chat:message:delta",
                                                        "data": {
                                                            "role": "assistant",
                                                            "content": content,
                                                        },
                                                    }
                                                )
                                else:
                                    # Handle plain text streaming (for smaller models)
                                    # Process line by line for plain text
                                    while "\n" in buffer:
                                        line, buffer = buffer.split("\n", 1)
                                        if line.strip():  # Only process non-empty lines
                                            self.log.debug(
                                                f"Processing plain text line: {repr(line[:100])}"
                                            )

                                            # Normalize content
                                            content = line.replace("\\n", "\n")
                                            n8n_response += content + "\n"

                                            # Emit delta for plain text
                                            if __event_emitter__:
                                                await __event_emitter__(
                                                    {
                                                        "type": "chat:message:delta",
                                                        "data": {
                                                            "role": "assistant",
                                                            "content": content + "\n",
                                                        },
                                                    }
                                                )

                            # Process any remaining content in buffer (CRITICAL FIX)
                            if buffer.strip():
                                self.log.debug(
                                    f"Processing remaining buffer content: {repr(buffer[:100])}"
                                )

                                # Try to extract from mixed content first
                                remaining_content = (
                                    self.extract_content_from_mixed_stream(buffer)
                                )

                                # If that doesn't work, use buffer as-is
                                if not remaining_content:
                                    remaining_content = buffer.strip()

                                if remaining_content:
                                    # Normalize escaped newlines to actual newlines (like non-streaming)
                                    remaining_content = remaining_content.replace(
                                        "\\n", "\n"
                                    )

                                    # Accumulate final buffer content
                                    n8n_response += remaining_content

                                    # Emit final buffer delta
                                    if __event_emitter__:
                                        await __event_emitter__(
                                            {
                                                "type": "chat:message:delta",
                                                "data": {
                                                    "role": "assistant",
                                                    "content": remaining_content,
                                                },
                                            }
                                        )

                            # NOW process all think blocks in the complete response
                            if n8n_response and "<think>" in n8n_response.lower():
                                # Use regex to find and replace all think blocks at once
                                think_pattern = re.compile(
                                    r"<think>\s*(.*?)\s*</think>",
                                    re.IGNORECASE | re.DOTALL,
                                )

                                think_counter = 0

                                def replace_think_block(match):
                                    nonlocal think_counter
                                    think_counter += 1
                                    thought_content = match.group(1).strip()
                                    if thought_content:
                                        completed_thoughts.append(thought_content)

                                        # Format each line with > for blockquote while preserving formatting
                                        quoted_lines = []
                                        for line in thought_content.split("\n"):
                                            quoted_lines.append(f"> {line}")
                                        quoted_content = "\n".join(quoted_lines)

                                        # Return details block with custom thought formatting
                                        return f"""<details>
<summary>Thought {think_counter}</summary>

{quoted_content}

</details>"""
                                    return ""

                                # Replace all think blocks with details blocks in the complete response
                                n8n_response = think_pattern.sub(
                                    replace_think_block, n8n_response
                                )

                            # ALWAYS emit final complete message (critical for UI update)
                            if __event_emitter__:
                                # Ensure we have some response to show
                                if not n8n_response.strip():
                                    n8n_response = "(Empty response received from N8N)"
                                    self.log.warning(
                                        "Empty response received from N8N, using fallback message"
                                    )

                                # Add tool calls section if present
                                if intermediate_steps:
                                    tool_calls_section = (
                                        self._format_tool_calls_section(
                                            intermediate_steps, for_streaming=False
                                        )
                                    )
                                    if tool_calls_section:
                                        n8n_response += tool_calls_section
                                        self.log.info(
                                            f"Added {len(intermediate_steps)} tool calls to response"
                                        )

                                await __event_emitter__(
                                    {
                                        "type": "chat:message",
                                        "data": {
                                            "role": "assistant",
                                            "content": n8n_response,
                                        },
                                    }
                                )
                                if completed_thoughts:
                                    # Clear any thinking status indicator
                                    await __event_emitter__(
                                        {
                                            "type": "status",
                                            "data": {
                                                "action": "thinking",
                                                "done": True,
                                                "hidden": True,
                                            },
                                        }
                                    )

                            self.log.info(
                                f"Streaming completed successfully. Total response length: {len(n8n_response)}"
                            )

                        except Exception as e:
                            self.log.error(f"Streaming error: {e}")

                            # In case of streaming errors, try to emit whatever we have
                            if n8n_response:
                                self.log.info(
                                    f"Emitting partial response due to error: {len(n8n_response)} chars"
                                )
                                if __event_emitter__:
                                    await __event_emitter__(
                                        {
                                            "type": "chat:message",
                                            "data": {
                                                "role": "assistant",
                                                "content": n8n_response,
                                            },
                                        }
                                    )
                            else:
                                # If no response was accumulated, provide error message
                                error_msg = f"Streaming error occurred: {str(e)}"
                                n8n_response = error_msg
                                if __event_emitter__:
                                    await __event_emitter__(
                                        {
                                            "type": "chat:message",
                                            "data": {
                                                "role": "assistant",
                                                "content": error_msg,
                                            },
                                        }
                                    )
                        finally:
                            await cleanup_response(response, session)

                        # Update conversation with response
                        body["messages"].append(
                            {"role": "assistant", "content": n8n_response}
                        )
                        await self.emit_simple_status(
                            __event_emitter__, "complete", "Streaming complete", True
                        )
                        return n8n_response
                    else:
                        # Fallback to non-streaming response (robust parsing)
                        self.log.info(
                            "Processing regular response from N8N (non-streaming)"
                        )

                        async def read_body_safely():
                            text_body = None
                            json_body = None
                            lowered = content_type.lower()
                            try:
                                # Read as text first (works for all content types)
                                text_body = await response.text()

                                # Try to parse as JSON regardless of content-type
                                # (N8N might return JSON with text/html content-type)
                                try:
                                    json_body = json.loads(text_body)
                                    self.log.debug(
                                        f"Successfully parsed response body as JSON (content-type was: {content_type})"
                                    )
                                except json.JSONDecodeError:
                                    # If it starts with [{ or { it might be JSON wrapped in something
                                    if text_body.strip().startswith(
                                        "[{"
                                    ) or text_body.strip().startswith("{"):
                                        self.log.warning(
                                            f"Response looks like JSON but failed to parse (content-type: {content_type})"
                                        )
                                    else:
                                        self.log.debug(
                                            f"Response is not JSON, will use as plain text (content-type: {content_type})"
                                        )
                            except Exception as e_inner:
                                self.log.error(
                                    f"Error reading response body: {e_inner}"
                                )
                            return json_body, text_body

                        response_json, response_text = await read_body_safely()
                        self.log.debug(f"Parsed JSON body: {response_json}")
                        if response_json is None and response_text:
                            snippet = (
                                (response_text[:300] + "...")
                                if len(response_text) > 300
                                else response_text
                            )
                            self.log.debug(f"Raw text body snippet: {snippet}")

                        # Extract intermediateSteps from non-streaming response
                        intermediate_steps = []
                        if isinstance(response_json, list):
                            # Handle array response format
                            self.log.debug(
                                f"Response is an array with {len(response_json)} items"
                            )
                            for item in response_json:
                                if (
                                    isinstance(item, dict)
                                    and "intermediateSteps" in item
                                ):
                                    steps = item.get("intermediateSteps", [])
                                    intermediate_steps.extend(steps)
                                    self.log.debug(
                                        f"Found {len(steps)} intermediate steps in array item"
                                    )
                        elif isinstance(response_json, dict):
                            # Handle single object response format
                            self.log.debug(
                                f"Response is a dict with keys: {list(response_json.keys())}"
                            )
                            intermediate_steps = response_json.get(
                                "intermediateSteps", []
                            )
                            if intermediate_steps:
                                self.log.debug(
                                    f"Found intermediateSteps field with {len(intermediate_steps)} items"
                                )
                        else:
                            self.log.debug(
                                f"Response is not JSON (type: {type(response_json)}), cannot extract intermediateSteps"
                            )

                        if intermediate_steps:
                            self.log.info(
                                f"✓ Found {len(intermediate_steps)} intermediate steps in non-streaming response"
                            )
                        else:
                            self.log.debug(
                                "No intermediate steps found in non-streaming response"
                            )

                        def extract_message(data) -> str:
                            if data is None:
                                return ""
                            if isinstance(data, dict):
                                # Prefer configured field
                                if self.valves.RESPONSE_FIELD in data and isinstance(
                                    data[self.valves.RESPONSE_FIELD], (str, list)
                                ):
                                    val = data[self.valves.RESPONSE_FIELD]
                                    if isinstance(val, list):
                                        return "\n".join(str(v) for v in val if v)
                                    return str(val)
                                # Common generic keys fallback
                                for key in (
                                    "content",
                                    "text",
                                    "output",
                                    "answer",
                                    "message",
                                ):
                                    if key in data and isinstance(
                                        data[key], (str, list)
                                    ):
                                        val = data[key]
                                        return (
                                            "\n".join(val)
                                            if isinstance(val, list)
                                            else str(val)
                                        )
                                # Flatten simple dict of scalars
                                try:
                                    flat = []
                                    for k, v in data.items():
                                        if isinstance(v, (str, int, float)):
                                            flat.append(f"{k}: {v}")
                                    return "\n".join(flat)
                                except Exception:
                                    return ""
                            if isinstance(data, list):
                                # Take first meaningful element
                                for item in data:
                                    m = extract_message(item)
                                    if m:
                                        return m
                                return ""
                            if isinstance(data, (str, int, float)):
                                return str(data)
                            return ""

                        n8n_response = extract_message(response_json)
                        if not n8n_response and response_text:
                            # Use raw text fallback (strip trailing whitespace only)
                            n8n_response = response_text.rstrip()

                        if not n8n_response:
                            n8n_response = (
                                "(Received empty response or unknown format from N8N)"
                            )

                        # Post-process for <think> blocks (non-streaming mode)
                        try:
                            if n8n_response and "<think>" in n8n_response.lower():
                                # First, normalize escaped newlines to actual newlines
                                normalized_response = n8n_response.replace("\\n", "\n")

                                # Use case-insensitive patterns to find and replace each think block
                                think_pattern = re.compile(
                                    r"<think>\s*(.*?)\s*</think>",
                                    re.IGNORECASE | re.DOTALL,
                                )

                                think_counter = 0

                                def replace_think_block(match):
                                    nonlocal think_counter
                                    think_counter += 1
                                    thought_content = match.group(1).strip()

                                    # Format each line with > for blockquote while preserving formatting
                                    quoted_lines = []
                                    for line in thought_content.split("\n"):
                                        quoted_lines.append(f"> {line}")
                                    quoted_content = "\n".join(quoted_lines)

                                    return f"""<details>
<summary>Thought {think_counter}</summary>

{quoted_content}

</details>"""

                                # Replace each <think>...</think> with its own details block
                                n8n_response = think_pattern.sub(
                                    replace_think_block, normalized_response
                                )
                        except Exception as post_e:
                            self.log.debug(
                                f"Non-streaming thinking parse failed: {post_e}"
                            )

                        # Add tool calls section if present (non-streaming mode)
                        if intermediate_steps:
                            tool_calls_section = self._format_tool_calls_section(
                                intermediate_steps, for_streaming=False
                            )
                            if tool_calls_section:
                                n8n_response += tool_calls_section
                                self.log.info(
                                    f"Added {len(intermediate_steps)} tool calls to non-streaming response"
                                )

                        # Extract token usage from N8N response (best-effort)
                        usage = None
                        if isinstance(response_json, dict):
                            usage = response_json.get("usage")
                        elif isinstance(response_json, list):
                            for item in response_json:
                                if isinstance(item, dict) and "usage" in item:
                                    usage = item["usage"]
                                    break

                        # Cleanup
                        await cleanup_response(response, session)
                        session = None

                        # Append assistant message
                        body["messages"].append(
                            {"role": "assistant", "content": n8n_response}
                        )

                        await self.emit_simple_status(
                            __event_emitter__, "complete", "Complete", True
                        )

                        # Return OpenAI-format dict with usage so the middleware saves it to DB
                        if usage:
                            return {
                                "choices": [
                                    {
                                        "message": {
                                            "role": "assistant",
                                            "content": n8n_response,
                                        }
                                    }
                                ],
                                "usage": usage,
                            }
                        return n8n_response

                else:
                    error_text = await response.text()
                    self.log.error(
                        f"N8N error: Status {response.status} - {error_text}"
                    )
                    await cleanup_response(response, session)

                    # Parse error message for better user experience
                    user_error_msg = f"N8N Error {response.status}"
                    try:
                        error_json = json.loads(error_text)
                        if "message" in error_json:
                            user_error_msg = f"N8N Error: {error_json['message']}"
                        if "hint" in error_json:
                            user_error_msg += f"\n\nHint: {error_json['hint']}"
                    except:
                        # If not JSON, use raw text but truncate if too long
                        if error_text:
                            truncated = (
                                error_text[:200] + "..."
                                if len(error_text) > 200
                                else error_text
                            )
                            user_error_msg = f"N8N Error {response.status}: {truncated}"

                    # Return error as chat message string
                    await self.emit_simple_status(
                        __event_emitter__, "error", user_error_msg, True
                    )
                    return user_error_msg

            except Exception as e:
                error_msg = f"Connection or processing error: {str(e)}"
                self.log.exception(error_msg)

                # Clean up session if it exists
                if session:
                    await session.close()

                # Return error as chat message string
                await self.emit_simple_status(
                    __event_emitter__,
                    "error",
                    error_msg,
                    True,
                )
                return error_msg

        # If no message is available alert user
        else:
            error_msg = "No messages found in the request body"
            self.log.warning(error_msg)
            await self.emit_simple_status(
                __event_emitter__,
                "error",
                error_msg,
                True,
            )
            return error_msg