Refactor swizzle_execute for async and sync handling

- Enhanced the swizzle_execute function to differentiate between asynchronous and synchronous execution paths based on the origin_execute function's nature. - Improved the structure of the swizzle_execute function to ensure proper handling of parameters and execution flow for both async and sync scenarios. - Maintained existing error handling while ensuring consistent behavior across execution types. These changes improve the flexibility and robustness of the execution handling in the application.
tweak: optional
2025-07-23 22:41:18 +08:00 · 2025-07-23 21:59:13 +08:00 · 2025-07-23 21:52:51 +08:00 · 2025-07-23 21:46:55 +08:00 · 2025-07-23 12:41:15 +08:00 · 2025-07-23 11:58:47 +08:00
8 changed files with 355 additions and 3643 deletions
@@ -2,8 +2,6 @@

 Open source comfyui deployment platform, a `vercel` for generative workflow infra. (serverless hosted gpu with vertical intergation with comfyui)

-Check out our latest lcoal demo -> https://github.com/comfy-deploy/comfyui-api-comfydeploy
-
 > [!NOTE]  
 > Im looking for creative hacker to join ComfyDeploy's core team! DM me on [twitter](https://x.com/BennyKokMusic)

@@ -283,45 +283,11 @@ async def post_prompt(json_data):
        prompt = json_data["prompt"]
        prompt_id = json_data.get("prompt_id") or str(uuid.uuid4())

-        partial_execution_targets = None
-        if "partial_execution_targets" in json_data:
-            partial_execution_targets = json_data["partial_execution_targets"]
-
-        # Handle different validate_prompt signatures (newest to oldest)
-        valid = None
-        last_error = None
-
-        # v0.3.48 (3 args)
        try:
-            valid = await execution.validate_prompt(
-                prompt_id, prompt, partial_execution_targets
-            )
+            valid = await execution.validate_prompt(prompt_id, prompt)
        except TypeError as e:
-            last_error = e
-            logger.debug(
-                f"validate_prompt with 3 params not supported, trying with 2. Debug: {last_error}"
-            )
-
-        # v0.3.45 - 0.3.47 (2 args)
-        if valid is None:
-            try:
-                valid = await execution.validate_prompt(prompt_id, prompt)
-            except TypeError as e:
-                last_error = e
-                logger.debug(
-                    f"validate_prompt with 2 params not supported, trying legacy signature. Debug: {last_error}"
-                )
-
-        # v0.3.44 or older (1 arg)
-        if valid is None:
-            try:
-                valid = execution.validate_prompt(prompt)
-            except TypeError as e:
-                last_error = e
-                logger.error(
-                    f"validate_prompt failed with all signatures. Last error: {last_error}"
-                )
-                raise
+            logger.warning(f"Trying old validate_prompt signature: {e}")
+            valid = execution.validate_prompt(prompt)

        extra_data = {}
        if "extra_data" in json_data:
@@ -579,6 +545,22 @@ async def comfy_deploy_run(request):
    data = await request.json()

    client_id = data.get("client_id")
+    # We proxy the request to Comfy Deploy, this is a native run
+    if "is_native_run" in data:
+        async with aiohttp.ClientSession() as session:
+            # pprint(data)
+            # headers = request.headers.copy()
+            # headers['Content-Type'] = 'application/json'
+            async with session.post(
+                data.get("native_run_api_endpoint"),
+                json=data,
+                headers={
+                    "Content-Type": "application/json",
+                    "Authorization": request.headers.get("Authorization"),
+                },
+            ) as response:
+                data = await response.json()
+                # print(data)

    if "cd_token" in data:
        token = data["cd_token"]
@@ -590,24 +572,6 @@ async def comfy_deploy_run(request):
            if len(parts) == 2 and parts[0].lower() == "bearer":
                token = parts[1]

-    # We proxy the request to Comfy Deploy, this is a native run
-    if "is_native_run" in data:
-        async with aiohttp.ClientSession() as session:
-            data.pop("cd_token", None)
-            # pprint(data)
-            # headers = request.headers.copy()
-            # headers['Content-Type'] = 'application/json'
-            async with session.post(
-                data.get("native_run_api_endpoint"),
-                json=data,
-                headers={
-                    "Content-Type": "application/json",
-                    "Authorization": "Bearer " + token,
-                },
-            ) as response:
-                data = await response.json()
-                # print(data)
-
    # In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
    workflow_api = data.get("workflow_api_raw")
    # The prompt id generated from comfy deploy, can be None
@@ -674,14 +638,6 @@ async def comfy_deploy_run(request):
    return web.json_response(res, status=status)


-@server.PromptServer.instance.routes.post("/comfyui-deploy/interrupt")
-async def interrupt_prompt(request):
-    data = await request.json()
-    prompt_id = data.get("prompt_id")
-    await update_run(prompt_id, Status.CANCELLED)
-    return web.json_response({"message": "Prompt interrupted"}, status=200)
-
-
 async def stream_prompt(data, token):
    # In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
    workflow_api = data.get("workflow_api_raw")
@@ -2142,7 +2098,6 @@ async def upload_in_background(
                ("model_file", "format", "application/octet-stream"),
                ("result", "format", "application/octet-stream"),
                ("text_file", "format", "text/plain"),
-                ("audio", "format", "audio/mpeg"),
            ]:
                items = data.get(file_type, [])

@@ -2230,7 +2185,6 @@ async def update_run_with_output(
            or "model_file" in data
            or "result" in data
            or "text_file" in data
-            or "audio" in data
        )
    if bypass_upload and have_upload_media:
        print(
@@ -3005,7 +2959,6 @@ async def create_workflow_proxy(request):
    name = data.get("name")
    workflow_json = data.get("workflow_json")
    workflow_api = data.get("workflow_api")
-    machine_id = data.get("machine_id")
    api_url = data.get("api_url", "https://api.comfydeploy.com")

    auth_header = request.headers.get("Authorization")
@@ -3025,7 +2978,6 @@ async def create_workflow_proxy(request):
        "name": name,
        "workflow_json": json.dumps(workflow_json),
        "workflow_api": json.dumps(workflow_api),
-        "machine_id": machine_id,
    }

    try:
@@ -3148,147 +3100,3 @@ async def get_workflow_proxy(request):
            return web.json_response(json_data, status=response.status)
    except Exception as e:
        return web.json_response({"error": str(e)}, status=500)
-
-
-# for getting a machine by id
-@server.PromptServer.instance.routes.get("/comfyui-deploy/machine")
-async def get_machine_proxy(request):
-    machine_id = request.rel_url.query.get("machine_id")
-    api_url = request.rel_url.query.get("api_url", "https://api.comfydeploy.com")
-    auth_header = request.headers.get("Authorization")
-
-    if not auth_header:
-        return web.json_response(
-            {"error": "Authorization header is required"}, status=401
-        )
-
-    target_url = f"{api_url}/api/machine/{machine_id}"
-
-    try:
-        await ensure_client_session()
-        async with client_session.get(
-            target_url, headers={"Authorization": auth_header}
-        ) as response:
-            json_data = await response.json()
-            return web.json_response(json_data, status=response.status)
-    except Exception as e:
-        return web.json_response({"error": str(e)}, status=500)
-
-
-# for fetching docker steps from current snapshot
-@server.PromptServer.instance.routes.post("/comfyui-deploy/snapshot-to-docker")
-async def snapshot_to_docker_proxy(request):
-    data = await request.json()
-    snapshot = data.get("snapshot")
-    api_url = data.get("api_url", "https://api.comfydeploy.com")
-    auth_header = request.headers.get("Authorization")
-
-    if not auth_header:
-        return web.json_response(
-            {"error": "Authorization header is required"}, status=401
-        )
-
-    target_url = f"{api_url}/api/snapshot-to-docker"
-
-    request_body = snapshot
-
-    try:
-        await ensure_client_session()
-        async with client_session.post(
-            target_url, json=request_body, headers={"Authorization": auth_header}
-        ) as response:
-            json_data = await response.json()
-            return web.json_response(json_data, status=response.status)
-    except Exception as e:
-        return web.json_response({"error": str(e)}, status=500)
-
-
-# update a serverless machine with machine id
-@server.PromptServer.instance.routes.post("/comfyui-deploy/machine/update")
-async def update_machine_proxy(request):
-    data = await request.json()
-    machine_id = data.get("machine_id")
-    comfyui_version = data.get("comfyui_version", None)
-    docker_steps = data.get("docker_steps")
-    api_url = data.get("api_url", "https://api.comfydeploy.com")
-    auth_header = request.headers.get("Authorization")
-
-    if not auth_header:
-        return web.json_response(
-            {"error": "Authorization header is required"}, status=401
-        )
-
-    target_url = f"{api_url}/api/machine/serverless/{machine_id}"
-
-    request_body = {"docker_command_steps": docker_steps}
-
-    if comfyui_version:
-        request_body["comfyui_version"] = comfyui_version
-
-    try:
-        await ensure_client_session()
-        async with client_session.patch(
-            target_url, json=request_body, headers={"Authorization": auth_header}
-        ) as response:
-            json_data = await response.json()
-            return web.json_response(json_data, status=response.status)
-    except Exception as e:
-        return web.json_response({"error": str(e)}, status=500)
-
-
-@server.PromptServer.instance.routes.post("/comfyui-deploy/machine/create")
-async def create_machine_proxy(request):
-    data = await request.json()
-    name = data.get("name")
-    docker_command_steps = data.get("docker_command_steps")
-    comfyui_version = data.get("comfyui_version")
-    api_url = data.get("api_url", "https://api.comfydeploy.com")
-    auth_header = request.headers.get("Authorization")
-
-    if not auth_header:
-        return web.json_response(
-            {"error": "Authorization header is required"}, status=401
-        )
-
-    target_url = f"{api_url}/api/machine/serverless"
-
-    request_body = {
-        "name": name,
-        "docker_command_steps": docker_command_steps,
-        "comfyui_version": comfyui_version,
-        "gpu": "A10G",
-    }
-
-    try:
-        await ensure_client_session()
-        async with client_session.post(
-            target_url, json=request_body, headers={"Authorization": auth_header}
-        ) as response:
-            json_data = await response.json()
-            return web.json_response(json_data, status=response.status)
-    except Exception as e:
-        return web.json_response({"error": str(e)}, status=500)
-
-
-# get latest comfyui version
-@server.PromptServer.instance.routes.get("/comfyui-deploy/comfyui-version")
-async def get_comfyui_version_proxy(request):
-    api_url = request.rel_url.query.get("api_url", "https://api.comfydeploy.com")
-    auth_header = request.headers.get("Authorization")
-
-    if not auth_header:
-        return web.json_response(
-            {"error": "Authorization header is required"}, status=401
-        )
-
-    target_url = f"{api_url}/api/latest-hashes"
-
-    try:
-        await ensure_client_session()
-        async with client_session.get(
-            target_url, headers={"Authorization": auth_header}
-        ) as response:
-            json_data = await response.json()
-            return web.json_response(json_data, status=response.status)
-    except Exception as e:
-        return web.json_response({"error": str(e)}, status=500)
@@ -18,7 +18,6 @@ class Status(Enum):
    SUCCESS = "success"
    FAILED = "failed"
    UPLOADING = "uploading"
-    CANCELLED = "cancelled"


 class StreamingPrompt(BaseModel):
@@ -1,7 +1,7 @@
 [project]
 name = "comfyui-deploy"
 description = "Open source comfyui deployment platform, a vercel for generative workflow infra."
-version = "2.3.4"
+version = "2.2.1"
 license = { file = "LICENSE" }
 dependencies = ["aiofiles", "pydantic", "opencv-python", "imageio-ffmpeg", "tabulate", "brotli"]

@@ -1,82 +0,0 @@
-// Snapshot Utilities
-// Centralized snapshot fetching with ComfyUI version fallback
-
-/**
- * Fetches the current snapshot with ComfyUI version fallback
- * If the snapshot response has null comfyui field, it will fetch the latest ComfyUI version
- * and update the snapshot with the comfyui_hash
- *
- * @param {Function} getDataFn - Function that returns { apiKey, apiUrl } for ComfyUI version API calls
- * @returns {Promise<Object>} - The snapshot data with comfyui field populated
- */
-export async function fetchSnapshot(getDataFn = null) {
-  try {
-    // Fetch the current snapshot
-    const response = await fetch("/snapshot/get_current");
-    if (!response.ok) {
-      throw new Error(`Snapshot fetch failed: ${response.status}`);
-    }
-
-    const snapshot = await response.json();
-
-    // Check if comfyui field is null and we have getDataFn for fallback
-    if (snapshot.comfyui === null && getDataFn) {
-      console.log(
-        "ComfyUI version is null in snapshot, fetching latest version..."
-      );
-
-      try {
-        const data = getDataFn();
-        if (data && data.apiKey) {
-          const comfyuiVersionResponse = await fetch(
-            `/comfyui-deploy/comfyui-version?api_url=${encodeURIComponent(
-              data.apiUrl || "https://api.comfydeploy.com"
-            )}`,
-            {
-              headers: {
-                Authorization: `Bearer ${data.apiKey}`,
-              },
-            }
-          );
-
-          if (comfyuiVersionResponse.ok) {
-            const versionData = await comfyuiVersionResponse.json();
-            if (versionData.comfyui_hash) {
-              console.log(
-                `Using ComfyUI hash from API: ${versionData.comfyui_hash}`
-              );
-              snapshot.comfyui = versionData.comfyui_hash;
-            }
-          } else {
-            console.warn(
-              "Failed to fetch ComfyUI version from API:",
-              comfyuiVersionResponse.status
-            );
-          }
-        }
-      } catch (error) {
-        console.warn("Error fetching ComfyUI version fallback:", error);
-        // Continue with original snapshot even if fallback fails
-      }
-    }
-
-    return snapshot;
-  } catch (error) {
-    console.error("Error fetching snapshot:", error);
-    throw error;
-  }
-}
-
-/**
- * Simple snapshot fetch without ComfyUI version fallback
- * Use this when you don't need the ComfyUI version fallback logic
- *
- * @returns {Promise<Object>} - The snapshot data as-is
- */
-export async function fetchSnapshotSimple() {
-  const response = await fetch("/snapshot/get_current");
-  if (!response.ok) {
-    throw new Error(`Snapshot fetch failed: ${response.status}`);
-  }
-  return response.json();
-}
@@ -49,7 +49,6 @@ async function fetchWorkflows(getData, offset = 0, limit = 20, search = "") {

 function createWorkflowItem(workflow, getTimeAgo, getData) {
  const li = document.createElement("li");
-  let loadingToast = null;
  li.style.cssText = `
      border-bottom: 1px solid #444;
      background: transparent;
@@ -75,7 +74,7 @@ function createWorkflowItem(workflow, getTimeAgo, getData) {
      }

      // Show loading toast
-      loadingToast = window.app.extensionManager.toast.add({
+      const loadingToast = window.app.extensionManager.toast.add({
        severity: "info",
        summary: "Loading workflow...",
        detail: `Loading "${workflow.name}"`,
@@ -109,33 +108,6 @@ function createWorkflowItem(workflow, getTimeAgo, getData) {
          // Load the workflow
          window.app.loadGraphData(latestVersion.workflow);

-          // Wait a bit for the graph to fully load before checking for ComfyDeploy node
-          await new Promise((resolve) => setTimeout(resolve, 100));
-
-          // Check if ComfyDeploy node exists, if not add it back
-          const graph = window.app.graph;
-          let deployMeta = graph.findNodesByType("ComfyDeploy");
-
-          if (deployMeta.length === 0) {
-            // Add ComfyDeploy node with workflow metadata
-            graph.beforeChange();
-            const node = LiteGraph.createNode("ComfyDeploy");
-            node.configure({
-              widgets_values: [
-                workflow.name, // workflow_name
-                workflow.id, // workflow_id
-                latestVersion.version, // version
-              ],
-            });
-            node.pos = [0, 0];
-            graph.add(node);
-            graph.afterChange();
-
-            console.log(
-              `Added ComfyDeploy node with: name="${workflow.name}", id="${workflow.id}", version="${latestVersion.version}"`
-            );
-          }
-
          // Show success toast
          window.app.extensionManager.toast.add({
            severity: "success",
@@ -155,9 +127,7 @@ function createWorkflowItem(workflow, getTimeAgo, getData) {
        life: 5000,
      });
    } finally {
-      if (loadingToast) {
-        loadingToast.close();
-      }
+      loadingToast.close();
    }
  });

@@ -309,7 +279,7 @@ async function initializeWorkflowsList(element, getData, getTimeAgo) {
        list-style-type: none;
        padding: 0;
        margin: 0;
-        height: calc(100vh - 550px);
+        height: calc(100vh - 350px);
        overflow-y: auto;
        scrollbar-width: thin;
        scrollbar-color: #666 transparent;
Author	SHA1	Message	Date
KarrixLee	761e6db573	Refactor swizzle_execute for async and sync handling - Enhanced the swizzle_execute function to differentiate between asynchronous and synchronous execution paths based on the origin_execute function's nature. - Improved the structure of the swizzle_execute function to ensure proper handling of parameters and execution flow for both async and sync scenarios. - Maintained existing error handling while ensuring consistent behavior across execution types. These changes improve the flexibility and robustness of the execution handling in the application.	2025-07-23 22:41:18 +08:00
KarrixLee	2e527f74b5	tweak: optional	2025-07-23 21:59:13 +08:00
KarrixLee	7d8f4563af	Refactor post_prompt and origin_execute for synchronous handling - Modified the post_prompt function to call validate_prompt synchronously when a TypeError occurs, improving error handling. - Updated the origin_execute function to execute synchronously, ensuring consistent behavior during execution. These changes enhance the robustness of prompt validation and execution processes in the application.	2025-07-23 21:52:51 +08:00
KarrixLee	fba00a3c4d	Enhance error handling in post_prompt and origin_execute functions - Added try-except blocks in post_prompt to handle TypeErrors during prompt validation, allowing for fallback to an older signature. - Implemented similar error handling in the origin_execute function to manage potential TypeErrors, ensuring robust execution flow. - Improved logging to capture issues with function signatures, aiding in debugging. These changes improve the resilience of the application when dealing with prompt and execution validation.	2025-07-23 21:46:55 +08:00
KarrixLee	7f56d18599	Refactor handle_execute to support asynchronous execution - Updated the swizzle_execute function to be asynchronous, allowing for non-blocking execution of the origin_execute function. - Added missing parameters for pending_async_nodes in the swizzle_execute function call. - Ensured that the result from origin_execute is awaited, improving the handling of asynchronous operations. These changes enhance the performance and responsiveness of the execution handling in the application.	2025-07-23 12:41:15 +08:00
KarrixLee	bdb2c1b85d	Refactor post_prompt and send_prompt functions for async handling - Updated the post_prompt function to be asynchronous, allowing for non-blocking execution when validating prompts. - Adjusted calls to post_prompt in send_prompt and comfy_deploy_run to await the asynchronous execution, ensuring proper handling of prompt submissions. - Commented out unused prompt construction code to streamline the function. These changes enhance the performance and responsiveness of prompt handling in the application.	2025-07-23 11:58:47 +08:00
KarrixLee	ee658f90b2	Refactor post_prompt function to include prompt_id handling - Updated the post_prompt function in custom_routes.py to retrieve and validate a prompt_id from the incoming JSON data, defaulting to a new UUID if not provided. - Adjusted the validation call to use prompt_id alongside the prompt, ensuring proper identification and processing of prompts. These changes enhance the functionality of prompt handling within the application.	2025-07-23 11:42:16 +08:00