generate_image
Create images from text descriptions using ComfyUI's AI generation capabilities. Specify what to include and exclude for customized visual outputs.
Instructions
Generate an image using ComfyUI
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| prompt | Yes | Positive prompt describing what you want in the image | |
| negative_prompt | No | Negative prompt describing what you don't want | bad hands, bad quality |
| seed | No | Seed for reproducible generation | |
| width | No | Image width in pixels | |
| height | No | Image height in pixels |
Implementation Reference
- src/comfy_ui_mcp_server/server.py:36-74 (registration)Registers the generate_image tool with the MCP server by returning it in the list_tools handler, including description and input schema.@self.app.list_tools() async def list_tools() -> List[Tool]: """List available image generation tools.""" return [ Tool( name="generate_image", description="Generate an image using ComfyUI", inputSchema={ "type": "object", "properties": { "prompt": { "type": "string", "description": "Positive prompt describing what you want in the image" }, "negative_prompt": { "type": "string", "description": "Negative prompt describing what you don't want", "default": "bad hands, bad quality" }, "seed": { "type": "number", "description": "Seed for reproducible generation", "default": 8566257 }, "width": { "type": "number", "description": "Image width in pixels", "default": 512 }, "height": { "type": "number", "description": "Image height in pixels", "default": 512 } }, "required": ["prompt"] } ) ]
- Executes the generate_image tool: builds ComfyUI workflow JSON, queues prompt via HTTP, listens on WebSocket for execution completion and binary image data.async def generate_image( self, prompt: str, negative_prompt: str, seed: int, width: int, height: int ) -> bytes: """Generate an image using ComfyUI.""" # Construct ComfyUI workflow workflow = { "4": { "class_type": "CheckpointLoaderSimple", "inputs": { "ckpt_name": "v1-5-pruned-emaonly.safetensors" } }, "5": { "class_type": "EmptyLatentImage", "inputs": { "batch_size": 1, "height": height, "width": width } }, "6": { "class_type": "CLIPTextEncode", "inputs": { "clip": ["4", 1], "text": prompt } }, "7": { "class_type": "CLIPTextEncode", "inputs": { "clip": ["4", 1], "text": negative_prompt } }, "3": { "class_type": "KSampler", "inputs": { "cfg": 8, "denoise": 1, "latent_image": ["5", 0], "model": ["4", 0], "negative": ["7", 0], "positive": ["6", 0], "sampler_name": "euler", "scheduler": "normal", "seed": seed, "steps": 20 } }, "8": { "class_type": "VAEDecode", "inputs": { "samples": ["3", 0], "vae": ["4", 2] } }, "save_image_websocket": { "class_type": "SaveImageWebsocket", "inputs": { "images": ["8", 0] } }, "save_image": { "class_type": "SaveImage", "inputs": { "images": ["8", 0], "filename_prefix": "mcp" } } } try: prompt_response = await self.queue_prompt(workflow) logger.info(f"Queued prompt, got response: {prompt_response}") prompt_id = prompt_response["prompt_id"] except Exception as e: logger.error(f"Error queuing prompt: {e}") raise uri = f"ws://{self.config.server_address}/ws?clientId={self.config.client_id}" logger.info(f"Connecting to websocket at {uri}") async with websockets.connect(uri) as websocket: while True: try: message = await websocket.recv() if isinstance(message, str): try: data = json.loads(message) logger.info(f"Received text message: {data}") if data.get("type") == "executing": exec_data = data.get("data", {}) if exec_data.get("prompt_id") == prompt_id: node = exec_data.get("node") logger.info(f"Processing node: {node}") if node is None: logger.info("Generation complete signal received") break except: pass else: logger.info(f"Received binary message of length: {len(message)}") if len(message) > 8: # Check if we have actual image data return message[8:] # Remove binary header else: logger.warning(f"Received short binary message: {message}") except websockets.exceptions.ConnectionClosed as e: logger.error(f"WebSocket connection closed: {e}") break except Exception as e: logger.error(f"Error processing message: {e}") continue raise RuntimeError("No valid image data received")
- Input schema defining parameters for the generate_image tool: prompt (required), negative_prompt, seed, width, height with defaults.inputSchema={ "type": "object", "properties": { "prompt": { "type": "string", "description": "Positive prompt describing what you want in the image" }, "negative_prompt": { "type": "string", "description": "Negative prompt describing what you don't want", "default": "bad hands, bad quality" }, "seed": { "type": "number", "description": "Seed for reproducible generation", "default": 8566257 }, "width": { "type": "number", "description": "Image width in pixels", "default": 512 }, "height": { "type": "number", "description": "Image height in pixels", "default": 512 } }, "required": ["prompt"] }
- Helper function to queue the ComfyUI workflow prompt via HTTP POST to /prompt endpoint and return the prompt_id.async def queue_prompt(self, prompt: Dict[str, Any]) -> Dict[str, Any]: """Queue a prompt with ComfyUI.""" async with aiohttp.ClientSession() as session: try: async with session.post( f"http://{self.config.server_address}/prompt", json={ "prompt": prompt, "client_id": self.config.client_id } ) as response: if response.status != 200: text = await response.text() raise RuntimeError(f"Failed to queue prompt: {response.status} - {text}") return await response.json() except aiohttp.ClientError as e: raise RuntimeError(f"HTTP request failed: {e}")