# Directory Structure
```
├── .dockerignore
├── .gitignore
├── .python-version
├── Dockerfile
├── Dockerfile.sse
├── LICENSE
├── pyproject.toml
├── README.md
├── requirements.txt
├── resources
│ └── run_workflow_from_file_demo.png
├── src
│ ├── .env
│ ├── client
│ │ ├── __init__.py
│ │ └── comfyui.py
│ ├── server.py
│ └── test_comfyui.py
├── uv.lock
└── workflows
└── text_to_image.json
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.dockerignore:
--------------------------------------------------------------------------------
```
test.py
test
# Git
.git
.gitignore
.github
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
venv/
ENV/
.env
.venv
# IDE
.idea/
.vscode/
*.swp
*.swo
.DS_Store
# Docker
Dockerfile
docker-compose.yml
.docker/
# Logs
*.log
logs/
# Test
test/
tests/
.coverage
.pytest_cache/
htmlcov/
# Documentation
docs/
*.md
```
--------------------------------------------------------------------------------
/src/.env:
--------------------------------------------------------------------------------
```
# if true, the server will return the url of the generated image
# if false, the server will return the image bytes
RETURN_URL=true
# ComfyUI parameters
# When using Docker, it's recommended to set the host as 'host.docker.internal'
# But, if you don't set up a separate volume, downloading images with the download_image tool will be difficult
# COMFYUI_HOST=host.docker.internal
COMFYUI_HOST=localhost
COMFYUI_PORT=8188
# If you use authentication setting, you can set the authentication type and value
# COMFYUI_AUTHENTICATION="Bearer <token>"
# COMFYUI_AUTHENTICATION="Basic <b64_encoded_credentials>"
# The directory of the workflow files
WORKFLOW_DIR=workflows
# MCP Transport
# sse or stdio
MCP_TRANSPORT=stdio
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
test
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
# .env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# ComfyUI MCP Server
## 1. Overview
- A server implementation for integrating ComfyUI with MCP.
- ⚠️ IMPORTANT: This server requires a running ComfyUI server.
- You must either host your own ComfyUI server,
- or have access to an existing ComfyUI server address.
<a href="https://glama.ai/mcp/servers/@Overseer66/comfyui-mcp-server">
<img width="380" height="200" src="https://glama.ai/mcp/servers/@Overseer66/comfyui-mcp-server/badge" alt="ComfyUI Server MCP server" />
</a>
---
## 2. Debugging
### 2.1 ComfyUI Debugging
```bash
python src/test_comfyui.py
```
### 2.2 MCP Debugging
```bash
mcp dev src/server.py
```
---
## 3. Installation and Configuration
### 3.1 ComfyUI Configuration
- Edit `src/.env` to set ComfyUI host and port:
```env
COMFYUI_HOST=localhost
COMFYUI_PORT=8188
```
### 3.2 Adding Custom Workflows
- To add new tools, place your workflow JSON files in the `workflows` directory and declare them as new tools in the system.
---
## 4. Built-in Tools
- **text_to_image**
- Returns only the URL of the generated image.
- To get the actual image:
- Use the `download_image` tool, or
- Access the URL directly in your browser.
- **download_image**
- Downloads images generated by other tools (like `text_to_image`) using the image URL.
- **run_workflow_with_file**
- Run a workflow by providing the path to a workflow JSON file.
```
# You should ask to agent like this.
Run comfyui workflow with text_to_image.json
```
- example image of CursorAI
- **run_workflow_with_json**
- Run a workflow by providing the workflow JSON data directly.
```
# You should ask to agent like this.
Run comfyui workflow with this
{
"3": {
"inputs": {
"seed": 156680208700286,
"steps": 20,
... (workflow JSON example)
}
```
---
## 5. How to Run
### 5.1 Using UV (Recommended)
- Example `mcp.json`:
```json
{
"mcpServers": {
"comfyui": {
"command": "uv",
"args": [
"--directory",
"PATH/MCP/comfyui",
"run",
"--with",
"mcp",
"--with",
"websocket-client",
"--with",
"python-dotenv",
"mcp",
"run",
"src/server.py:mcp"
]
}
}
}
```
### 5.2 Using Docker
- Downloading images to a local folder with `download_image` may be difficult since the Docker container does not share the host filesystem.
- When using Docker, consider:
1. Set `RETURN_URL=false` in `.env` to receive image data as bytes.
2. Set `COMFYUI_HOST` in `.env` to the appropriate address (e.g., `host.docker.internal` or your server's IP).
3. Note: Large image payloads may exceed response limits when using binary data.
#### 5.2.1 Build Docker Image
```bash
# First build image
docker image build -t mcp/comfyui .
```
```json
{
"mcpServers": {
"comfyui": {
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"-p",
"3001:3000",
"mcp/comfyui"
]
}
}
}
```
#### 5.2.2 Using Existing Images
Also you can use prebuilt image.
```json
{
"mcpServers": {
"comfyui": {
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"-p",
"3001:3000",
"overseer66/mcp-comfyui"
]
}
}
}
```
#### 5.2.3 Using SSE Transport
1. Run the SSE server with Docker:
```bash
docker run -i --rm -p 8001:8000 overseer66/mcp-comfyui-sse
```
2. Configure `mcp.json` (change localhost to your IP or domain if needed):
```json
{
"mcpServers": {
"comfyui": {
"url": "http://localhost:8001/sse"
}
}
}
```
> NOTE: When adding new workflows as tools, you need to rebuild and redeploy the Docker images to make them available.
---
```
--------------------------------------------------------------------------------
/src/client/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp[cli]
websocket-client
python-dotenv
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "comfyui-mcp-server"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"mcp[cli]>=1.6.0",
"python-dotenv>=1.1.0",
"websocket-client>=1.8.0",
]
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
FROM python:3.12-slim
WORKDIR /app
RUN apt-get update && apt-get install -y curl
COPY . .
RUN curl -LsSf https://astral.sh/uv/install.sh | sh
ENV PATH="/root/.local/bin:${PATH}"
RUN uv venv .venv && uv pip install -r pyproject.toml
EXPOSE 3000
ENTRYPOINT ["uv", "run", "--with", "mcp", "mcp", "run", "src/server.py:mcp"]
```
--------------------------------------------------------------------------------
/workflows/text_to_image.json:
--------------------------------------------------------------------------------
```json
{
"3": {
"inputs": {
"seed": 156680208700286,
"steps": 20,
"cfg": 8,
"sampler_name": "euler",
"scheduler": "normal",
"denoise": 1,
"model": [
"4",
0
],
"positive": [
"6",
0
],
"negative": [
"7",
0
],
"latent_image": [
"5",
0
]
},
"class_type": "KSampler",
"_meta": {
"title": "KSampler"
}
},
"4": {
"inputs": {
"ckpt_name": "v1-5-pruned-emaonly-fp16.safetensors"
},
"class_type": "CheckpointLoaderSimple",
"_meta": {
"title": "Load Checkpoint"
}
},
"5": {
"inputs": {
"width": 512,
"height": 512,
"batch_size": 1
},
"class_type": "EmptyLatentImage",
"_meta": {
"title": "Empty Latent Image"
}
},
"6": {
"inputs": {
"text": "beautiful scenery nature glass bottle landscape, , purple galaxy bottle,",
"clip": [
"4",
1
]
},
"class_type": "CLIPTextEncode",
"_meta": {
"title": "CLIP Text Encode (Prompt)"
}
},
"7": {
"inputs": {
"text": "text, watermark",
"clip": [
"4",
1
]
},
"class_type": "CLIPTextEncode",
"_meta": {
"title": "CLIP Text Encode (Prompt)"
}
},
"8": {
"inputs": {
"samples": [
"3",
0
],
"vae": [
"4",
2
]
},
"class_type": "VAEDecode",
"_meta": {
"title": "VAE Decode"
}
},
"9": {
"inputs": {
"filename_prefix": "ComfyUI",
"images": [
"8",
0
]
},
"class_type": "SaveImage",
"_meta": {
"title": "Save Image"
}
}
}
```
--------------------------------------------------------------------------------
/src/server.py:
--------------------------------------------------------------------------------
```python
import os
import json
import urllib.request
import urllib.parse
from typing import Any
from client.comfyui import ComfyUI
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
load_dotenv()
mcp = FastMCP("comfyui")
@mcp.tool()
async def text_to_image(prompt: str, seed: int, steps: int, cfg: float, denoise: float) -> Any:
"""Generate an image from a prompt.
Args:
prompt: The prompt to generate the image from.
seed: The seed to use for the image generation.
steps: The number of steps to use for the image generation.
cfg: The CFG scale to use for the image generation.
denoise: The denoise strength to use for the image generation.
"""
auth = os.environ.get("COMFYUI_AUTHENTICATION")
comfy = ComfyUI(
url=f'http://{os.environ.get("COMFYUI_HOST", "localhost")}:{os.environ.get("COMFYUI_PORT", 8188)}',
authentication=auth
)
images = await comfy.process_workflow("text_to_image", {"prompt": prompt, "seed": seed, "steps": steps, "cfg": cfg, "denoise": denoise}, return_url=os.environ.get("RETURN_URL", "true").lower() == "true")
return images
@mcp.tool()
async def download_image(url: str, save_path: str) -> Any:
"""Download an image from a URL and save it to a file.
Args:
url: The URL of the image to download.
save_path: The absolute path to save the image to. Must be an absolute path, otherwise the image will be saved relative to the server location.
"""
urllib.request.urlretrieve(url, save_path)
return {"success": True}
@mcp.tool()
async def run_workflow_from_file(file_path: str) -> Any:
"""Run a workflow from a file.
Args:
file_path: The absolute path to the file to run.
"""
with open(file_path, "r", encoding="utf-8") as f:
workflow = json.load(f)
auth = os.environ.get("COMFYUI_AUTHENTICATION")
comfy = ComfyUI(
url=f'http://{os.environ.get("COMFYUI_HOST", "localhost")}:{os.environ.get("COMFYUI_PORT", 8188)}',
authentication=auth
)
images = await comfy.process_workflow(workflow, {}, return_url=os.environ.get("RETURN_URL", "true").lower() == "true")
return images
@mcp.tool()
async def run_workflow_from_json(json_data: dict) -> Any:
"""Run a workflow from a JSON data.
Args:
json_data: The JSON data to run.
"""
workflow = json_data
auth = os.environ.get("COMFYUI_AUTHENTICATION")
comfy = ComfyUI(
url=f'http://{os.environ.get("COMFYUI_HOST", "localhost")}:{os.environ.get("COMFYUI_PORT", 8188)}',
authentication=auth
)
images = await comfy.process_workflow(workflow, {}, return_url=os.environ.get("RETURN_URL", "true").lower() == "true")
return images
if __name__ == "__main__":
mcp.run(transport=os.environ.get("MCP_TRANSPORT", "stdio"))
```
--------------------------------------------------------------------------------
/src/client/comfyui.py:
--------------------------------------------------------------------------------
```python
import os
import websocket
import json
import uuid
import urllib.parse
import urllib.request
from typing import Dict, Any
class ComfyUI:
def __init__(self, url: str, authentication: str = None):
self.url = url
self.authentication = authentication
self.client_id = str(uuid.uuid4())
self.headers = {
"Content-Type": "application/json"
}
if authentication:
self.headers["Authorization"] = authentication
def get_image(self, filename, subfolder, folder_type):
data = {"filename": filename, "subfolder": subfolder, "type": folder_type}
url_values = urllib.parse.urlencode(data)
url = f"{self.url}/view?{url_values}"
req = urllib.request.Request(url, headers=self.headers)
with urllib.request.urlopen(req) as response:
return response.read()
def get_history(self, prompt_id):
url = f"{self.url}/history/{prompt_id}"
req = urllib.request.Request(url, headers=self.headers)
with urllib.request.urlopen(req) as response:
return json.loads(response.read())
def queue_prompt(self, prompt):
p = {"prompt": prompt, "client_id": self.client_id}
data = json.dumps(p).encode("utf-8")
req = urllib.request.Request(
f"{self.url}/prompt",
headers=self.headers,
data=data
)
return json.loads(urllib.request.urlopen(req).read())
async def process_workflow(self, workflow: Any, params: Dict[str, Any], return_url: bool = False):
if isinstance(workflow, str):
workflow_path = os.path.join(os.environ.get("WORKFLOW_DIR", "workflows"), f"{workflow}.json")
if not os.path.exists(workflow_path):
raise Exception(f"Workflow {workflow} not found")
with open(workflow_path, "r", encoding='utf-8') as f:
prompt = json.load(f)
else:
prompt = workflow
self.update_workflow_params(prompt, params)
ws = websocket.WebSocket()
ws_url = f"ws://{os.environ.get("COMFYUI_HOST", "localhost")}:{os.environ.get("COMFYUI_PORT", 8188)}/ws?clientId={self.client_id}"
if self.authentication:
ws.connect(ws_url, header=[f"Authorization: {self.authentication}"])
else:
ws.connect(ws_url)
try:
images = self.get_images(ws, prompt, return_url)
return images
finally:
ws.close()
def get_images(self, ws, prompt, return_url):
prompt_id = self.queue_prompt(prompt)["prompt_id"]
output_images = {}
while True:
out = ws.recv()
if isinstance(out, str):
message = json.loads(out)
if message["type"] == "executing":
data = message["data"]
if data["node"] is None and data["prompt_id"] == prompt_id:
break
else:
continue
history = self.get_history(prompt_id)[prompt_id]
for node_id in history["outputs"]:
node_output = history["outputs"][node_id]
if "images" in node_output:
if return_url:
output_images[node_id] = []
for image in node_output["images"]:
data = {"filename": image["filename"], "subfolder": image["subfolder"], "type": image["type"]}
url_values = urllib.parse.urlencode(data)
url = f"{self.url}/view?{url_values}"
output_images[node_id].append(url)
else:
output_images[node_id] = [
self.get_image(image["filename"], image["subfolder"], image["type"])
for image in node_output["images"]
]
return output_images
def update_workflow_params(self, prompt, params):
if not params:
return
for node in prompt.values():
if node["class_type"] == "CLIPTextEncode" and "text" in params:
if isinstance(node["inputs"]["text"], str):
node["inputs"]["text"] = params["text"]
elif node["class_type"] == "KSampler":
if "seed" in params:
node["inputs"]["seed"] = params["seed"]
if "steps" in params:
node["inputs"]["steps"] = params["steps"]
if "cfg" in params:
node["inputs"]["cfg"] = params["cfg"]
if "denoise" in params:
node["inputs"]["denoise"] = params["denoise"]
elif node["class_type"] == "LoadImage" and "image" in params:
node["inputs"]["image"] = params["image"]
```