workflow: use anyio.to_thread to send sync server functions to thread pool (#13612)

* run server funcs in thread pool with asyncio

* add changeset

* lint

* add changeset

* rm import

* use anyio + add test

* limit call_fn concurrency

* format

* test fix

* refactor

* add changeset

* clean up

* format + lint

* typefix

* move import

* tweak

* rm direct call_fn tests

* rm call_fn server function

* add sanitized name error

* lint

* format

* fix test

* tweak

* format

* add changeset

* add flag for unicode

* add blocks check

* use gr.api() to register bound fn endpoints

* save lock

* trigger job.cancel() on abort

* prevent race in image editor

* format

* format

* error tweak

* revert

* wrap bound fn exceptions as gr.Error

* format

---------

Co-authored-by: gradio-pr-bot <gradio-pr-bot@users.noreply.github.com>

Author: hannahblair

.changeset/funny-nights-leave.md

@@ -0,0 +1,6 @@
+---
+"@gradio/workflowcanvas": patch
+"gradio": patch
+---
+
+fix:workflow: use anyio.to_thread to send sync server functions to thread pool

gradio/routes.py

@@ -27,6 +27,7 @@
     cast,
 )
 
+import anyio
 import fastapi
 import httpx
 import markupsafe
@@ -1691,7 +1692,9 @@ async def component_server(
             if inspect.iscoroutinefunction(fn):
                 return await fn(*processed_input)
             else:
-                return fn(*processed_input)
+                return await anyio.to_thread.run_sync(
+                    fn, *processed_input, limiter=app.get_blocks().limiter
+                )
 
         @router.get(
             "/queue/status",

gradio/workflow.py

@@ -10,6 +10,7 @@
 import secrets
 import sys
 import tempfile
+import threading
 import types
 import urllib.parse
 import warnings
@@ -18,13 +19,15 @@
 from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING, Optional, TypedDict, Union, get_type_hints
 
+import anyio
 import httpx
 from huggingface_hub import HfApi
 from huggingface_hub import get_token as hf_get_token
 
 import gradio as gr
 from gradio.blocks import Blocks
 from gradio.components.workflowcanvas import WorkflowCanvas
+from gradio.context import Context
 from gradio.helpers import special_args as _special_args
 from gradio.oauth import OAuthProfile, OAuthToken
 from gradio.route_utils import Request
@@ -67,7 +70,7 @@ class _CuratedCache(TypedDict):
 
 
 _CURATED_CACHE: _CuratedCache = {"fetched_at": 0.0, "items": None}
-_CURATED_LOCK = __import__("threading").Lock()
+_CURATED_LOCK = threading.Lock()
 
 
 def _bundled_snapshot_path() -> str:
@@ -135,6 +138,7 @@ def _load_curated() -> list[dict]:
 # Scalar-only — everything else (str, list, dict, custom classes) falls through
 # to the default "text" port type, which round-trips as JSON.
 _PY_TO_PORT = {int: "number", float: "number", bool: "boolean"}
+_SANITIZE_RE = re.compile(r"[^a-zA-Z0-9_-]")
 
 
 def _build_edges(
@@ -1304,6 +1308,11 @@ def __init__(
         self._bound: dict[str, Callable] = bind or {}
         self._edges: list[tuple[str, str]] = edges or []
 
+        if Context.root_block is not None:
+            raise ValueError(
+                "gr.Workflow cannot be created inside another gr.Blocks context."
+            )
+
         warnings.warn(
             "gr.Workflow is currently in beta. Its API and UX may change in future releases.",
             UserWarning,
@@ -1338,11 +1347,10 @@ def _load_initial() -> str | None:
                 return None
 
         bound = self._bound
+        _save_lock = threading.Lock()
 
         def call_fn(
-            data,
-            _request: Optional[Request] = None,
-            _token: Optional[OAuthToken] = None,
+            data, request: Optional[Request] = None, token: Optional[OAuthToken] = None
         ) -> str:
             fn_name = data[0] if data else ""
             try:
@@ -1359,10 +1367,15 @@ def call_fn(
                 args = json.loads(args_json)
                 if not isinstance(args, list):
                     args = [args]
-                args, *_ = _special_args(fn, args, _request, None, token=_token)
-                result = fn(*args)
-                result = list(result) if isinstance(result, (list, tuple)) else [result]
-                return json.dumps(result)
+                args, *_ = _special_args(fn, args, request, None, token=token)
+                if inspect.iscoroutinefunction(fn):
+                    import asyncio
+
+                    result = asyncio.run(fn(*args))
+                else:
+                    result = fn(*args)
+                out = list(result) if isinstance(result, (list, tuple)) else [result]
+                return json.dumps(out)
             except Exception as e:
                 logger.error("call_fn failed for %s: %s", fn_name, e, exc_info=True)
                 return json.dumps(
@@ -1459,17 +1472,17 @@ def save_workflow(
                     WorkflowGraph(parsed)
                 except ValueError as exc:
                     return json.dumps({"error": f"Invalid workflow schema: {exc}"})
-                with open(workflow_file, "w", encoding="utf-8") as f:
-                    f.write(payload)
-                # Re-derive API endpoints so /info + /call track the saved graph
-                # (outputs added / removed / renamed / retyped).
-                if self._api_endpoints is not None:
-                    try:
-                        self._api_endpoints.sync()
-                    except Exception:
-                        logger.error(
-                            "Workflow: endpoint sync after save failed", exc_info=True
-                        )
+                with _save_lock:
+                    with open(workflow_file, "w", encoding="utf-8") as f:
+                        f.write(payload)
+                    if self._api_endpoints is not None:
+                        try:
+                            self._api_endpoints.sync()
+                        except Exception:
+                            logger.error(
+                                "Workflow: endpoint sync after save failed",
+                                exc_info=True,
+                            )
                 return "ok"
             except Exception as e:
                 logger.error("save_workflow failed: %s", e, exc_info=True)
@@ -1491,7 +1504,6 @@ def save_workflow(
             curated_modalities,
             curated_modality_tasks,
             get_dataset_schema,
-            call_fn,
             list_bound_fns,
             get_workflow_api,
             save_workflow,
@@ -1520,6 +1532,51 @@ def _current_graph() -> WorkflowGraph | None:
                 server_functions=server_functions,
             )
 
+        def _wrap_bound_fn(fn: Callable) -> Callable:
+            async def wrapper(
+                args_json: str,
+                _request: Optional[Request] = None,
+                _token: Optional[OAuthToken] = None,
+            ) -> str:
+                args = json.loads(args_json)
+                if not isinstance(args, list):
+                    args = [args]
+                args, *_ = _special_args(fn, args, _request, None, token=_token)
+                try:
+                    result = (
+                        await fn(*args)
+                        if inspect.iscoroutinefunction(fn)
+                        else await anyio.to_thread.run_sync(
+                            lambda: fn(*args), limiter=self.limiter
+                        )
+                    )
+                except Exception as e:
+                    raise gr.Error(str(e)) from e
+                out = list(result) if isinstance(result, (list, tuple)) else [result]
+                return json.dumps(out)
+
+            return wrapper
+
+        if bound:
+            from gradio.workflow_api import _active_blocks
+
+            if len(bound) != len({_SANITIZE_RE.sub("_", n) for n in bound}):
+                sanitized = [_SANITIZE_RE.sub("_", n) for n in bound]
+                dupes = {s for s in sanitized if sanitized.count(s) > 1}
+                raise ValueError(
+                    f"gr.Workflow: bound function names produce duplicate endpoint "
+                    f"names after sanitizing: {dupes}. Rename one."
+                )
+
+            with _active_blocks(self):
+                for fn_name, fn in bound.items():
+                    sanitized_name = _SANITIZE_RE.sub("_", fn_name)
+                    gr.api(
+                        _wrap_bound_fn(fn),
+                        api_name=f"predict_fn_{sanitized_name}",
+                        concurrency_limit="default",
+                        api_visibility="undocumented",
+                    )
         # Expose each subject (output) as a named API endpoint reusing /info +
         # /call. The manager re-syncs on every save_workflow, so adding,
         # removing, renaming, or retyping an output updates the live API.

js/workflowcanvas/Index.svelte

@@ -45,7 +45,11 @@
 </script>
 
 <div class="workflow-fullscreen">
-	<WorkflowCanvas server={serverObj} {initialValue} />
+	<WorkflowCanvas
+		server={serverObj}
+		{initialValue}
+		gradio_shared={gradio.shared}
+	/>
 </div>
 
 <style>

js/workflowcanvas/workflow/WorkflowCanvas.svelte

@@ -76,8 +76,15 @@
 
 	let {
 		server = {},
-		initialValue = null
-	}: { server?: Record<string, any>; initialValue?: string | null } = $props();
+		initialValue = null,
+		gradio_shared = undefined
+	}: {
+		server?: Record<string, any>;
+		initialValue?: string | null;
+		gradio_shared?: Record<string, any> | undefined;
+	} = $props();
+
+	const gradio_client = $derived(gradio_shared?.client);
 
 	const auth = createHFAuth(() => server);
 
@@ -1529,9 +1536,35 @@
 					])
 			: undefined;
 
-		const callFnWithToken = server?.call_fn
-			? async (fnName: string, argsJson: string) =>
-					server.call_fn([fnName, argsJson])
+		const callFnWithToken = gradio_client
+			? async (fnName: string, argsJson: string) => {
+					const safeN = fnName.replace(/[^a-zA-Z0-9_-]/gu, "_");
+					const job = gradio_client.submit(`/predict_fn_${safeN}`, [argsJson]);
+					abortController?.signal.addEventListener(
+						"abort",
+						() => job.cancel(),
+						{
+							once: true
+						}
+					);
+					for await (const msg of job) {
+						if (msg.type === "data") {
+							return (msg.data as unknown[])[0] as string;
+						}
+						if (msg.type === "status" && msg.stage === "error") {
+							return JSON.stringify({
+								error: msg.message ?? "Function call failed",
+								error_type: "unknown",
+								suggestion: ""
+							});
+						}
+					}
+					return JSON.stringify({
+						error: "No data received from fn endpoint",
+						error_type: "unknown",
+						suggestion: ""
+					});
+				}
 			: undefined;
 
 		await executeWorkflow(

test/test_workflow.py

@@ -1,13 +1,12 @@
 import json
 import os
 import tempfile
-from typing import Optional
 
 import pytest
 
 import gradio as gr
 import gradio.workflow as workflow_module
-from gradio.oauth import OAuthProfile, OAuthToken
+from gradio.oauth import OAuthToken
 from gradio.route_utils import Request
 from gradio.workflow import (
     WRITE_TOKEN,
@@ -441,58 +440,12 @@ def text_generation(self, *args, **kwargs):
 
 
 class TestCallFn:
-    def _call_fn(self, tmp_path, bind):
-        wf = Workflow(graph=str(tmp_path / "wf.json"), bind=bind)
-        canvas = next(
-            b for b in wf.blocks.values() if b.get_block_name() == "workflowcanvas"
-        )
-        return canvas.call_fn
-
-    def test_calls_bound_fn(self, tmp_path):
-        call_fn = self._call_fn(tmp_path, {"echo": lambda x: x})
-        result = json.loads(call_fn(["echo", '["hello"]']))
-        assert result == ["hello"]
-
-    def test_injects_oauth_token(self, tmp_path):
-        received = {}
-
-        def fn_with_token(text: str, token: Optional[OAuthToken]) -> str:
-            received["token"] = token
-            return text
-
-        call_fn = self._call_fn(tmp_path, {"fn_with_token": fn_with_token})
-
-        class _MockRequest:
-            session = {
-                "oauth_info": {
-                    "access_token": "test-tok",
-                    "scope": "openid",
-                    "expires_at": 9999999999,
-                }
-            }
-
-        result = json.loads(
-            call_fn(["fn_with_token", '["hi"]'], _request=_MockRequest())
-        )
-        assert result == ["hi"]
-        assert received["token"].token == "test-tok"
-
-    def test_injects_direct_token_without_session(self, tmp_path):
-        def fn(text: str, token: OAuthToken, profile: Optional[OAuthProfile]) -> str:
-            return f"{token.token},{profile}"
-
-        call_fn = self._call_fn(tmp_path, {"fn": fn})
-        direct = OAuthToken.__new__(OAuthToken)
-        direct.token = "direct-token"
-        direct.scope = "openid"
-        direct.expires_at = 9999999999
-        result = json.loads(call_fn(["fn", '["hello"]'], _request=None, _token=direct))
-        assert result == ["direct-token,None"]
-
-    def test_unknown_fn_returns_error(self, tmp_path):
-        call_fn = self._call_fn(tmp_path, {"echo": lambda x: x})
-        result = json.loads(call_fn(["missing", "[]"]))
-        assert result.get("error_type") == "unknown"
+    def test_queue_endpoint_registered_for_bound_fn(self, tmp_path):
+        wf = Workflow(graph=str(tmp_path / "wf.json"), bind={"my_fn": lambda x: x})
+        api_names = [
+            fn.api_name for fn in wf.fns.values() if isinstance(fn.api_name, str)
+        ]
+        assert "predict_fn_my_fn" in api_names
 
 
 class TestCallSpaceValidation: