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) ```