### Pipeline Lifecycle Hooks: on_startup() and on_shutdown() in Python Source: https://context7.com/lalanikarim/comfy-mcp-pipeline/llms.txt These Python methods define lifecycle hooks for the pipeline. `on_startup()` installs the 'uv' package manager using pip, while `on_shutdown()` serves as a placeholder for cleanup operations during pipeline termination. Both methods print messages indicating their execution. ```python import subprocess import sys async def on_startup(self): """Called when the pipeline starts - installs uv package manager""" print(f"on_startup:{__name__}") subprocess.check_call([ sys.executable, "-m", "pip", "install", "uv", ]) async def on_shutdown(self): """Called when the pipeline shuts down - cleanup hook""" print(f"on_shutdown:{__name__}") ``` -------------------------------- ### Process Assistant Responses with outlet() in Python Source: https://context7.com/lalanikarim/comfy-mcp-pipeline/llms.txt The outlet() method in Python processes assistant responses after image generation. It converts base64 image data into file attachments for inline display in Open WebUI. This method expects the message content to start with 'data:' to identify image data. ```python async def outlet(self, body: dict, user: dict) -> dict: messages = body["messages"] last_message = messages[-1] # Check if the last assistant message contains base64 image data if (last_message["role"] == "assistant" and last_message["content"][:5] == "data:"): image_url = last_message["content"] content = messages[-2]["content"] # Get the original prompt # Replace content with description and attach image as file last_message["content"] = f"Generated: {content}" last_message["files"] = [{"type": "image", "url": image_url}] messages[-1] = last_message body["messages"] = messages return body ``` -------------------------------- ### Image Generation using MCP Client Source: https://context7.com/lalanikarim/comfy-mcp-pipeline/llms.txt Implements the `pipe()` method to generate images by interacting with the `comfy-mcp-server`. It handles prompt length limits, configures server parameters, and processes responses, returning either base64 encoded image data or a URL. ```python from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.types import ImageContent from typing import Generator, Iterator, Union, List import asyncio import os def pipe( self, user_message: str, model_id: str, messages: List[dict], body: dict ) -> Union[str, Generator, Iterator]: # Limit prompt length to 500 characters if len(user_message) > 500: return "" # Configure MCP server parameters with ComfyUI settings server_params = StdioServerParameters( command="/usr/local/bin/uvx", args=["comfy-mcp-server"], env={ "COMFY_URL": self.valves.COMFY_URL, "COMFY_URL_EXTERNAL": self.valves.COMFY_URL_EXTERNAL, "COMFY_WORKFLOW_JSON_FILE": self.valves.COMFY_WORKFLOW_JSON_FILE, "PROMPT_NODE_ID": self.valves.PROMPT_NODE_ID, "OUTPUT_NODE_ID": self.valves.OUTPUT_NODE_ID, "OUTPUT_MODE": "url", "PATH": os.getenv("PATH"), } ) async def apipe(user_message: str, model_id: str, messages: List[dict], body: dict): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() return await session.call_tool( "generate_image", arguments={"prompt": user_message} ) # Execute async function in appropriate event loop context try: loop = asyncio.get_running_loop() result = loop.run_until_complete(apipe(user_message, model_id, messages, body)) except RuntimeError: result = asyncio.run(apipe(user_message, model_id, messages, body)) # Process result - handle both image content and URL responses content = result.content[0] if isinstance(content, ImageContent): return f"data:{content.mimeType};base64, {content.data}" else: if content.text[:4] == 'http' and content.text[-11:] == 'type=output': return f"\n![image]({content.text})\n" return f"{content.text}\n" ``` -------------------------------- ### ComfyUI Flux Workflow Configuration in JSON Source: https://context7.com/lalanikarim/comfy-mcp-pipeline/llms.txt This JSON configuration defines a ComfyUI image generation workflow using a Flux-based approach. It includes components for CLIP text encoding, VAE decoding, and custom sampling parameters, specifying node types and their interconnections. ```json { "6": { "inputs": { "text": "tileset, pixalated, 2d, sprites", "clip": ["11", 0] }, "class_type": "CLIPTextEncode", "_meta": { "title": "CLIP Text Encode (Positive Prompt)" } }, "9": { "inputs": { "filename_prefix": "ComfyUI", "images": ["8", 0] }, "class_type": "SaveImage", "_meta": { "title": "Save Image" } }, "11": { "inputs": { "clip_name1": "t5xxl_fp8_e4m3fn.safetensors", "clip_name2": "clip_l.safetensors", "type": "flux", "device": "default" }, "class_type": "DualCLIPLoader", "_meta": { "title": "DualCLIPLoader" } }, "12": { "inputs": { "unet_name": "flux1-dev.safetensors", "weight_dtype": "default" }, "class_type": "UNETLoader", "_meta": { "title": "Load Diffusion Model" } }, "17": { "inputs": { "scheduler": "simple", "steps": 20, "denoise": 1, "model": ["30", 0] }, "class_type": "BasicScheduler", "_meta": { "title": "BasicScheduler" } }, "27": { "inputs": { "width": 200, "height": 200, "batch_size": 1 }, "class_type": "EmptySD3LatentImage", "_meta": { "title": "EmptySD3LatentImage" } } } ``` -------------------------------- ### Pipeline Configuration with Pydantic Source: https://context7.com/lalanikarim/comfy-mcp-pipeline/llms.txt Defines the `Pipeline` class and its nested `Valves` model for configuring connection settings between Open WebUI and ComfyUI. It uses Pydantic for data validation and defaults to environment variables or localhost settings. ```python from pydantic import BaseModel import os class Pipeline: class Valves(BaseModel): COMFY_URL: str # Internal URL for ComfyUI server COMFY_URL_EXTERNAL: str # External URL for ComfyUI server (for image URLs) COMFY_WORKFLOW_JSON_FILE: str # Path to the workflow JSON file PROMPT_NODE_ID: str # Node ID for text prompt in workflow OUTPUT_NODE_ID: str # Node ID for generated image output def __init__(self): self.name = "Comfy MCP Pipeline" self.valves = self.Valves( **{ "COMFY_URL": os.getenv("COMFY_URL", "http://localhost:8188"), "COMFY_URL_EXTERNAL": os.getenv("COMFY_URL_EXTERNAL", "http://localhost:8188"), "COMFY_WORKFLOW_JSON_FILE": os.getenv("COMFY_WORKFLOW_JSON_FILE", "/workflows/flux.json"), "PROMPT_NODE_ID": os.getenv("PROMPT_NODE_ID", "6"), "OUTPUT_NODE_ID": os.getenv("OUTPUT_NODE_ID", "9"), } ) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.