""" title: n8n Pipe Function author: Cole Medin author_url: https://www.youtube.com/@ColeMedin version: 0.2.1 This module defines a Pipe class that utilizes N8N for an Agent """ from typing import Optional, Callable, Awaitable from pydantic import BaseModel, Field import os import time import requests import logging import json # Configure logging logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def extract_event_info(event_emitter) -> tuple[Optional[str], Optional[str]]: logger.debug("Extracting event info from event_emitter") if not event_emitter or not event_emitter.__closure__: logger.debug("No event_emitter or closure found") 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") logger.debug(f"Extracted chat_id: {chat_id}, message_id: {message_id}") return chat_id, message_id logger.debug("No request_info found in closure") return None, None class Pipe: class Valves(BaseModel): n8n_url: str = Field(default="http://n8n:5678/webhook/invoke_n8n_agent") n8n_bearer_token: str = Field(default="...") input_field: str = Field(default="chatInput") response_field: str = Field(default="output") emit_interval: float = Field( default=2.0, description="Interval in seconds between status emissions" ) enable_status_indicator: bool = Field( default=True, description="Enable or disable status indicator emissions" ) def __init__(self): self.type = "pipe" self.id = "n8n_pipe" self.name = "N8N Pipe" self.valves = self.Valves() self.last_emit_time = 0 logger.info(f"N8N Pipe initialized with URL: {self.valves.n8n_url}") async def emit_status( self, __event_emitter__: Callable[[dict], Awaitable[None]], level: str, message: str, done: bool, ): current_time = time.time() logger.debug(f"emit_status called: level={level}, message={message}, done={done}") if ( __event_emitter__ and self.valves.enable_status_indicator and ( current_time - self.last_emit_time >= self.valves.emit_interval or done ) ): status_data = { "type": "status", "data": { "status": "complete" if done else "in_progress", "level": level, "description": message, "done": done, }, } logger.debug(f"Emitting status: {status_data}") await __event_emitter__(status_data) self.last_emit_time = current_time else: logger.debug("Status emission skipped (disabled or interval not reached)") async def pipe( self, body: dict, __user__: Optional[dict] = None, __event_emitter__: Callable[[dict], Awaitable[None]] = None, __event_call__: Callable[[dict], Awaitable[dict]] = None, ) -> Optional[dict]: logger.info("=== N8N Pipe execution started ===") logger.debug(f"Input body: {json.dumps(body, indent=2)}") logger.debug(f"User info: {__user__}") await self.emit_status( __event_emitter__, "info", "/Calling N8N Workflow...", False ) # Extract chat information chat_id, message_id = extract_event_info(__event_emitter__) logger.info(f"Chat context - ID: {chat_id}, Message ID: {message_id}") messages = body.get("messages", []) logger.info(f"Found {len(messages)} messages in request") # Verify a message is available if messages: question = messages[-1]["content"] logger.info(f"Processing question: {question[:100]}..." if len(question) > 100 else f"Processing question: {question}") try: # Prepare request headers = { "Authorization": f"Bearer {self.valves.n8n_bearer_token}", "Content-Type": "application/json", } payload = {"sessionId": f"{chat_id}"} payload[self.valves.input_field] = question logger.info(f"N8N Request Details:") logger.info(f" URL: {self.valves.n8n_url}") logger.info(f" Headers: {headers}") logger.info(f" Payload: {json.dumps(payload, indent=2)}") # Make request to N8N logger.info("Making HTTP request to N8N...") response = requests.post( self.valves.n8n_url, json=payload, headers=headers, timeout=30 # Add timeout ) logger.info(f"N8N Response:") logger.info(f" Status Code: {response.status_code}") logger.info(f" Headers: {dict(response.headers)}") logger.info(f" Raw Response: {response.text}") if response.status_code == 200: try: response_json = response.json() logger.debug(f"Parsed JSON response: {json.dumps(response_json, indent=2)}") if self.valves.response_field in response_json: n8n_response = response_json[self.valves.response_field] logger.info(f"Extracted {self.valves.response_field}: {n8n_response}") else: logger.error(f"Response field '{self.valves.response_field}' not found in response") logger.error(f"Available fields: {list(response_json.keys())}") raise Exception(f"Response field '{self.valves.response_field}' not found in N8N response") except json.JSONDecodeError as e: logger.error(f"Failed to parse JSON response: {e}") logger.error(f"Raw response text: {response.text}") raise Exception(f"Invalid JSON response from N8N: {e}") else: error_msg = f"N8N returned error status {response.status_code}: {response.text}" logger.error(error_msg) raise Exception(error_msg) # Add response to messages logger.info("Adding N8N response to messages") body["messages"].append({"role": "assistant", "content": n8n_response}) except requests.exceptions.ConnectionError as e: error_msg = f"Connection error to N8N: {str(e)}" logger.error(error_msg) await self.emit_status( __event_emitter__, "error", f"Error during sequence execution: {error_msg}", True, ) return {"error": error_msg} except requests.exceptions.Timeout as e: error_msg = f"Timeout connecting to N8N: {str(e)}" logger.error(error_msg) await self.emit_status( __event_emitter__, "error", f"Error during sequence execution: {error_msg}", True, ) return {"error": error_msg} except Exception as e: error_msg = f"Unexpected error: {str(e)}" logger.error(error_msg, exc_info=True) # Include full traceback await self.emit_status( __event_emitter__, "error", f"Error during sequence execution: {error_msg}", True, ) return {"error": error_msg} # If no message is available alert user else: error_msg = "No messages found in the request body" logger.warning(error_msg) await self.emit_status( __event_emitter__, "error", error_msg, True, ) body["messages"].append( { "role": "assistant", "content": error_msg, } ) await self.emit_status(__event_emitter__, "info", "Complete", True) logger.info("=== N8N Pipe execution completed ===") # Log final return value if 'n8n_response' in locals(): logger.info(f"Returning N8N response: {n8n_response}") return n8n_response else: logger.info("Returning None (no valid response)") return None