@guillaume-bieler/viewcomfy-api-rules

You are an expert in Python, FastAPI integrations and web app development. You are tasked with helping integrate the ViewComfy API into web applications using Python.

The ViewComfy API is a serverless API built using the FastAPI framework that can run custom ComfyUI workflows. The Python version makes requests using the httpx library,

When implementing the API, remember that the first time you call it, you might experience a cold start. Moreover, generation times can vary between workflows; some might be less than 2 seconds, while some might take several minutes.

When calling the API, the params object can't be empty. If nothing else is specified, change the seed value.

The data comes back from the API with the following format: { "prompt_id": "string", # Unique identifier for the prompt "status": "string", # Current execution status "completed": bool, # Whether execution is complete "execution_time_seconds": float, # Time taken to execute "prompt": dict, # Original prompt configuration "outputs": [ # List of output files (optional) { "filename": "string", # Name of the output file "content_type": "string", # MIME type of the file "data": "string", # Base64 encoded file content "size": int # File size in bytes }, # ... potentially multiple output files ] }

ViewComfy documentation:

================================================
FILE: other_resources/guide_to_setting_up_and_using_ViewComfy_API.md
================================================
Deploying your workflow
The first thing you will need to do is to deploy your ComfyUI workflow on your ViewComfy dashboard using the workflow_api.json file.

Calling the workflow with the API
The ViewComfy API is a REST API that can be called with a standard POST request but also supports streaming responses via Server-Sent Events. This second option allows for real-time tracking of the ComfyUI logs.

Getting your API keys
In order to use your API endpoint, you will first need to create your API keys from the ViewComfy dashboard.

2. Extracting your workflow parameters

Before setting up the request is to identify the parameters in your workflow. This is done by using ViewComfy_API/Python/workflow_parameters_maker.py from the example API code to flatten your workflow_api.json.


The flattened json file should look like this:

{
"_3-node-class_type-info": "KSampler",
"3-inputs-cfg": 6,

…  

"_6-node-class_type-info": "CLIP Text Encode (Positive Prompt)",  
"6-inputs-clip": [  
    "38",  
    0  
],  
"6-inputs-text": "A woman raising her head with hair blowing in the wind",  

…  

"_52-node-class_type-info": "Load Image",  
"52-inputs-image": "<path_to_my_image>",  

…  
}


This dictionary contains all the parameters in your workflow. The key for each parameter contains the node id from your workflow_api.json file, whether it is an input, and the parameter’s input name. Keys that start with “_” are just there to give you context on the node corresponding to id, they are not parameters.

In this example, the first key-value pair shows that node 3 is the KSampler and that “3-inputs-cfg” sets its corresponding cfg value.

**3. Updating the script with your parameter**

First thing to do is to copy the ViewComfy endpoint from your dashboard and set it to view_comfy_api_url. You should also get the “Client ID” and “Client Secret” you made earlier, and set the client_id and client_secret values:

view_comfy_api_url = "<Your_ViewComfy_endpoint>"
client_id = "<Your_ViewComfy_client_id>"
client_secret = "<Your_ViewComfy_client_secret>"


You can then set the parameters using the keys from the json file you created in the previous step. In this example, we will change the prompt and the input image:

params = {}
params["6-inputs-text"] = "A flamingo dancing on top of a server in a pink universe, masterpiece, best quality, very aesthetic"
params["52-inputs-image"] = open("/home/gbieler/GitHub/API_tests/input_img.png", "rb")


**4. Calling the API**

Once you are done adding your parameters to ViewComfy_API/Python/main.py, you can call the API by running:

python main.py


This will send your parameters to ViewComfy_API/Python/api.py where all the functions to call the API and handle the outputs are stored.

By default the script runs the “infer_with_logs” function which returns the generation logs from ComfyUI via a streaming response. If you would rather call the API via a standard POST request, you can use “infer” instead.

The result object returned by the API will contain the workflow outputs as well as the generation details.

Your outputs will automatically be saved in your working directory.

================================================
FILE: ViewComfy_API/README.MD
================================================
# ViewComfy API Example

## API

All the functions to call the API and handle the responses are in the api file (api.py). The main file (main.py) takes in the parameters that are specific from your workflow and in most cases will be the only file you need to edit.

#### The API file has two endpoints:

- infer: classic request-response endpoint where you wait for your request to finish before getting results back. 

- infer_with_logs: receives real-time updates with the ComfyUI logs (eg. progress bar). To make use of this endpoint, you need to pass a function that will be called each time a log message is received.

The endpoints can also take a workflow_api.json as a parameter. This is useful if you want to run a different workflow than the one you used when deploying.

### Get your API parameters

To extract all the parameters from your workflow_api.json, you can run the workflow_api_parameter_creator function. This will create a dictionary with all of the parameters inside the workflow.

\`\`\`python

python workflow_parameters_maker.py --workflow_api_path "<Path to your workflow_api.json file>"

Running the example
Install the dependencies:


pip install -r requirements.txt

Add your endpoint and set your API keys:

Change the view_comfy_api_url value inside main.py to the ViewComfy endpoint from your ViewComfy Dashboard. Do the same with the "client_id" and "client_secret" values using your API keys (you can also get them from your dashboard). If you want, you can change the parameters of the workflow inside main.py at the same time.

Call the API:


python main.py

Using the API with a different workflow
You can overwrite the default workflow_api.json when sending a request. Be careful if you need to install new node packs to run the new workflow. Having too many custom node packages can create some issues between the Python packages. This can increase ComfyUI start up time and in some cases break the ComfyUI installation.

To use an updated workflow (that works with your deployment) with the API, you can send the new workflow_api.json as a parameter by changing the override_workflow_api_path value. For example, using python:

override_workflow_api_path = "<path_to_your_new_workflow_api_file>"
================================================ FILE: ViewComfy_API/example_workflow/workflow_api(example).json
{ "3": { "inputs": { "seed": 268261030599666, "steps": 20, "cfg": 6, "sampler_name": "uni_pc", "scheduler": "simple", "denoise": 1, "model": [ "56", 0 ], "positive": [ "50", 0 ], "negative": [ "50", 1 ], "latent_image": [ "50", 2 ] }, "class_type": "KSampler", "_meta": { "title": "KSampler" } }, "6": { "inputs": { "text": "A flamingo dancing on top of a server in a pink universe, masterpiece, best quality, very aesthetic", "clip": [ "38", 0 ] }, "class_type": "CLIPTextEncode", "_meta": { "title": "CLIP Text Encode (Positive Prompt)" } }, "7": { "inputs": { "text": "Overexposure, static, blurred details, subtitles, paintings, pictures, still, overall gray, worst quality, low quality, JPEG compression residue, ugly, mutilated, redundant fingers, poorly painted hands, poorly painted faces, deformed, disfigured, deformed limbs, fused fingers, cluttered background, three legs, a lot of people in the background, upside down", "clip": [ "38", 0 ] }, "class_type": "CLIPTextEncode", "_meta": { "title": "CLIP Text Encode (Negative Prompt)" } },

...

"52": { "inputs": { "image": "SMT54Y6XHY1977QPBESY72WSR0.jpeg", "upload": "image" }, "class_type": "LoadImage", "_meta": { "title": "Load Image" } },

...

}

================================================ FILE: ViewComfy_API/Python/api.py
import json from io import BufferedReader from typing import Any, Callable, Dict, List import httpx

class FileOutput: """Represents a file output with its content encoded in base64"""

def __init__(self, filename: str, content_type: str, data: str, size: int):
    """
    Initialize a FileOutput object.

    Args:
        filename (str): Name of the output file
        content_type (str): MIME type of the file
        data (str): Base64 encoded file content
        size (int): Size of the file in bytes
    """
    self.filename = filename
    self.content_type = content_type
    self.data = data
    self.size = size
class PromptResult: def init( self, prompt_id: str, status: str, completed: bool, execution_time_seconds: float, prompt: Dict, outputs: List[Dict] | None = None, ): """ Initialize a PromptResult object.

    Args:
        prompt_id (str): Unique identifier for the prompt
        status (str): Current status of the prompt execution
        completed (bool): Whether the prompt execution is complete
        execution_time_seconds (float): Time taken to execute the prompt
        prompt (Dict): The original prompt configuration
        outputs (List[Dict], optional): List of output file data. Defaults to empty list.
    """
    self.prompt_id = prompt_id
    self.status = status
    self.completed = completed
    self.execution_time_seconds = execution_time_seconds
    self.prompt = prompt

    # Initialize outputs as FileOutput objects
    self.outputs = []
    if outputs:
        for output_data in outputs:
            self.outputs.append(
                FileOutput(
                    filename=output_data.get("filename", ""),
                    content_type=output_data.get("content_type", ""),
                    data=output_data.get("data", ""),
                    size=output_data.get("size", 0),
                )
            )
class ComfyAPIClient: def init( self, *, infer_url: str | None = None, client_id: str | None = None, client_secret: str | None = None, ): """ Initialize the ComfyAPI client with the server URL.

    Args:
        base_url (str): The base URL of the API server
    """
    if infer_url is None:
        raise Exception("infer_url is required")
    self.infer_url = infer_url

    if client_id is None:
        raise Exception("client_id is required")

    if client_secret is None:
        raise Exception("client_secret is required")

    self.client_id = client_id
    self.client_secret = client_secret

async def infer(
    self,
    *,
    data: Dict[str, Any],
    files: list[tuple[str, BufferedReader]] = [],
) -> Dict[str, Any]:
    """
    Make a POST request to the /api/infer-files endpoint with files encoded in form data.

    Args:
        data: Dictionary of form fields (logs, params, etc.)
        files: Dictionary mapping file keys to tuples of (filename, content, content_type)
               Example: {"composition_image": ("image.jpg", file_content, "image/jpeg")}

    Returns:
        Dict[str, Any]: Response from the server
    """

    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(
                self.infer_url,
                data=data,
                files=files,
                timeout=httpx.Timeout(2400.0),
                follow_redirects=True,
                headers={
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                },
            )

            if response.status_code == 201:
                return response.json()
            else:
                error_text = response.text
                raise Exception(
                    f"API request failed with status {response.status_code}: {error_text}"
                )
        except httpx.HTTPError as e:
            raise Exception(f"Connection error: {str(e)}")
        except Exception as e:
            raise Exception(f"Error during API call: {str(e)}")

async def consume_event_source(
    self, *, response, logging_callback: Callable[[str], None]
) -> Dict[str, Any] | None:
    """
    Process a streaming Server-Sent Events (SSE) response.

    Args:
        response: An active httpx streaming response object

    Returns:
        List of parsed event objects
    """
    current_data = ""
    current_event = "message"  # Default event type

    prompt_result = None
    # Process the response as it streams in
    async for line in response.aiter_lines():
        line = line.strip()
        if prompt_result:
            break
        # Empty line signals the end of an event
        if not line:
            if current_data:
                try:
                    if current_event in ["log_message", "error"]:
                        logging_callback(f"{current_event}: {current_data}")
                    elif current_event == "prompt_result":
                        prompt_result = json.loads(current_data)
                    else:
                        print(
                            f"Unknown event: {current_event}, data: {current_data}"
                        )
                except json.JSONDecodeError as e:
                    print("Invalid JSON: ...")
                    print(e)
                # Reset for next event
                current_data = ""
                current_event = "message"
            continue

        # Parse SSE fields
        if line.startswith("event:"):
            current_event = line[6:].strip()
        elif line.startswith("data:"):
            current_data = line[5:].strip()
        elif line.startswith("id:"):
            # Handle event ID if needed
            pass
        elif line.startswith("retry:"):
            # Handle retry directive if needed
            pass
    return prompt_result

async def infer_with_logs(
    self,
    *,
    data: Dict[str, Any],
    logging_callback: Callable[[str], None],
    files: list[tuple[str, BufferedReader]] = [],
) -> Dict[str, Any] | None:
    if data.get("logs") is not True:
        raise Exception("Set the logs to True for streaming the process logs")

    async with httpx.AsyncClient() as client:
        try:
            async with client.stream(
                "POST",
                self.infer_url,
                data=data,
                files=files,
                timeout=24000,
                follow_redirects=True,
                headers={
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                },
            ) as response:
                if response.status_code == 201:
                    # Check if it's actually a server-sent event stream
                    if "text/event-stream" in response.headers.get(
                        "content-type", ""
                    ):
                        prompt_result = await self.consume_event_source(
                            response=response, logging_callback=logging_callback
                        )
                        return prompt_result
                    else:
                        # For non-SSE responses, read the content normally
                        raise Exception(
                            "Set the logs to True for streaming the process logs"
                        )
                else:
                    error_response = await response.aread()
                    error_data = json.loads(error_response)
                    raise Exception(
                        f"API request failed with status {response.status_code}: {error_data}"
                    )
        except Exception as e:
            raise Exception(f"Error with streaming request: {str(e)}")
def parse_parameters(params: dict): """ Parse parameters from a dictionary to a format suitable for the API call.

Args:
    params (dict): Dictionary of parameters

Returns:
    dict: Parsed parameters
"""
parsed_params = {}
files = []
for key, value in params.items():
    if isinstance(value, BufferedReader):
        files.append((key, value))
    else:
        parsed_params[key] = value
return parsed_params, files
async def infer( *, params: Dict[str, Any], api_url: str, override_workflow_api: Dict[str, Any] | None = None, client_id: str, client_secret: str, ): """ Make an inference with real-time logs from the execution prompt

Args:
    api_url (str): The URL to send the request to
    params (dict): The parameter to send to the workflow
    override_workflow_api (dict): Optional override the default workflow_api of the deployment

Returns:
    PromptResult: The result of the inference containing outputs and execution details
"""
client = ComfyAPIClient(
    infer_url=api_url,
    client_id=client_id,
    client_secret=client_secret,
)

params_parsed, files = parse_parameters(params)
data = {
    "logs": False,
    "params": json.dumps(params_parsed),
    "workflow_api": json.dumps(override_workflow_api)
    if override_workflow_api
    else None,
}

# Make the API call
result = await client.infer(data=data, files=files)

return PromptResult(**result)
async def infer_with_logs( *, params: Dict[str, Any], logging_callback: Callable[[str], None], api_url: str, override_workflow_api: Dict[str, Any] | None = None, client_id: str, client_secret: str, ): """ Make an inference with real-time logs from the execution prompt

Args:
    api_url (str): The URL to send the request to
    params (dict): The parameter to send to the workflow
    override_workflow_api (dict): Optional override the default workflow_api of the deployment
    logging_callback (Callable[[str], None]): The callback function to handle logging messages

Returns:
    PromptResult: The result of the inference containing outputs and execution details
"""

client = ComfyAPIClient(
    infer_url=api_url,
    client_id=client_id,
    client_secret=client_secret,
)

params_parsed, files = parse_parameters(params)
data = {
    "logs": True,
    "params": json.dumps(params_parsed),
    "workflow_api": json.dumps(override_workflow_api)
    if override_workflow_api
    else None,
}

# Make the API call
result = await client.infer_with_logs(
    data=data,
    files=files,
    logging_callback=logging_callback,
)

if result:
    return PromptResult(**result)
\`\`\`
FILE: ViewComfy_API/Python/main.py
\`\`\`python
import asyncio import base64 import json import os from api import infer, infer_with_logs

async def api_examples():

view_comfy_api_url = "<Your_ViewComfy_endpoint>"
client_id = "<Your_ViewComfy_client_id>"
client_secret = "<Your_ViewComfy_client_secret>"

override_workflow_api_path = None # Advanced feature: overwrite default workflow with a new one

# Set parameters
params = {}

params["6-inputs-text"] = "A cat sorcerer"
params["52-inputs-image"] = open("input_folder/input_img.png", "rb")

override_workflow_api = None
if  override_workflow_api_path:
    if os.path.exists(override_workflow_api_path):  
        with open(override_workflow_api_path, "r") as f:
            override_workflow_api = json.load(f)
    else:
        print(f"Error: {override_workflow_api_path} does not exist")

def logging_callback(log_message: str):
    print(log_message)

# Call the API and wait for the results
# try:
#     prompt_result = await infer(
#         api_url=view_comfy_api_url,
#         params=params,
#         client_id=client_id,
#         client_secret=client_secret,
#     )
# except Exception as e:
#     print("something went wrong calling the api")
#     print(f"Error: {e}")
#     return

# Call the API and get the logs of the execution in real time
# you can use any function that you want
try:
    prompt_result = await infer_with_logs(
        api_url=view_comfy_api_url,
        params=params,
        logging_callback=logging_callback,
        client_id=client_id,
        client_secret=client_secret,
        override_workflow_api=override_workflow_api,
    )
except Exception as e:
    print("something went wrong calling the api")
    print(f"Error: {e}")
    return

if not prompt_result:
    print("No prompt_result generated")
    return

for file in prompt_result.outputs:
    try:
        # Decode the base64 data before writing to file
        binary_data = base64.b64decode(file.data)
        with open(file.filename, "wb") as f:
            f.write(binary_data)
        print(f"Successfully saved {file.filename}")
    except Exception as e:
        print(f"Error saving {file.filename}: {str(e)}")
if name == "main": asyncio.run(api_examples())
\`\`\`

================================================ 
FILE: ViewComfy_API/Python/requirements.txt
\`\`\`
httpx==0.28.1
\`\`\`

================================================ 
FILE: ViewComfy_API/Python/workflow_api_parameter_creator.py
\`\`\`python
from typing import Dict, Any

def workflow_api_parameters_creator(workflow: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: """ Flattens the workflow API JSON structure into a simple key-value object

Args:
    workflow: The workflow API JSON object

Returns:
    A flattened object with keys in the format "nodeId-inputs-paramName" or "nodeId-class_type-info"
"""
flattened: Dict[str, Any] = {}

# Iterate through each node in the workflow
for node_id, node in workflow.items():
    # Add the class_type-info key, preferring _meta.title if available
    class_type_info = node.get("_meta", {}).get("title") or node.get("class_type")
    flattened[f"_{node_id}-node-class_type-info"] = class_type_info
    
    # Process all inputs
    if "inputs" in node:
        for input_key, input_value in node["inputs"].items():
            flattened[f"{node_id}-inputs-{input_key}"] = input_value

return flattened
""" Example usage:

import json

with open('workflow_api.json', 'r') as f: workflow_json = json.load(f)

flattened = create_workflow_api_parameters(workflow_json) print(flattened) """
\`\`\`

================================================ 
FILE: ViewComfy_API/Python/workflow_parameters_maker.py
\`\`\`python
import json from workflow_api_parameter_creator import workflow_api_parameters_creator import argparse

parser = argparse.ArgumentParser(description='Process workflow API parameters') parser.add_argument('--workflow_api_path', type=str, required=True, help='Path to the workflow API JSON file')

Parse arguments
args = parser.parse_args()

with open(args.workflow_api_path, 'r') as f: workflow_json = json.load(f)

parameters = workflow_api_parameters_creator(workflow_json)

with open('workflow_api_parameters.json', 'w') as f: json.dump(parameters, f, indent=4)
\`\`\`