feat: add models.run on both the sync and async clients

CHANGELOG.md

@@ -12,7 +12,17 @@ notes for each version.
 
 ## [Unreleased]
 
-_Nothing yet — add an entry here when your change lands._
+### Added
+
+- `client.models.run(model, arguments)` on both `Comfy` and `AsyncComfy` — one
+  call that returns the completed generation. Where the platform has to
+  submit-and-poll an upstream provider, that polling happens server side inside
+  the call, so the client contract stays a single request. The result is the
+  provider's native payload, returned as-is rather than wrapped. The awaitable
+  form is `AsyncComfy`, not a `run_async()` suffix — there is deliberately no
+  suffixed variant, and a test asserts its absence. Runs use their own
+  10-minute timeout (the client's default is sized for ordinary API calls) and
+  send an `Idempotency-Key` on every call.
 
 ## [0.1.8] - 2026-08-13
 

README.md

@@ -276,6 +276,36 @@ namespace, and nothing extra is imported or constructed for it:
 `base_url` and `timeout` are a read-only view of that shared configuration;
 model operations are added to this namespace as they land.
 
+### `models.run` — one call, one result
+
+```python
+result = client.models.run("acme/flux/dev", {"prompt": "a cat", "steps": 4})
+result["images"][0]["url"]
+```
+
+`run` returns when the generation is **complete**. There is no submit step and
+nothing to poll: where the platform has to submit-and-poll an upstream
+provider, that happens server side inside this one call. The value you get back
+is the provider's own payload — decoded JSON, handed over as-is, with no
+wrapper class between you and the fields the provider documented.
+
+The awaitable form is the **async client**, not a differently-named method:
+
+```python
+async with AsyncComfy(api_key="comfyui-...") as client:
+    result = await client.models.run("acme/flux/dev", {"prompt": "a cat"})
+```
+
+There is no `run_async()`, and there will not be one — one operation, one name,
+and `await` is what makes it asynchronous.
+
+Because the server may legitimately hold the connection for minutes, `run` uses
+its own 10-minute timeout rather than the client's (which is sized for ordinary
+API calls). Pass `timeout=` seconds, an `httpx.Timeout`, or `None` to wait
+indefinitely. Each call also sends a fresh `Idempotency-Key`, so an accidental
+exact resend is rejected by the server instead of billing a second generation;
+pass `idempotency_key=` to choose the value yourself.
+
 ## Sync and async
 
 `Comfy` and `AsyncComfy` expose the identical surface — swap the import and

src/comfy_low/transport.py

@@ -10,6 +10,12 @@
 * **per-request timeout / abort** — every method takes ``timeout`` and the raw
   httpx cancellation applies.
 
+One binding is *not* backed by an ``operationId``: ``post_model_run``. The
+vendored contract declares no model routes yet, so it is hand-written against
+the agreed wire shape, kept out of ``comfy_low.OPERATION_IDS``, and confined to
+``model_run_request`` / ``_MODEL_RUN_PATH`` so vendoring the real route later is
+a one-place change.
+
 This layer contains no orchestration, retries, hashing, or reconnection — those
 live in ``comfy_sdk``.
 """
@@ -20,7 +26,7 @@
 import platform
 import secrets
 import sys
-from collections.abc import AsyncIterator, Iterator
+from collections.abc import AsyncIterator, Iterator, Mapping
 from contextlib import asynccontextmanager, contextmanager
 from datetime import datetime, timedelta, timezone
 from importlib.metadata import PackageNotFoundError
@@ -46,9 +52,47 @@
 # blocking iter_lines() forever. (Pass timeout=None to opt out of the timeout.)
 _SSE_IDLE_TIMEOUT = httpx.Timeout(10.0, read=45.0)
 
+#: Default timeout for a model run. A run is *awaited server-side*: the server
+#: holds the connection until the generation is complete, polling the upstream
+#: provider itself when that provider is submit/poll. So the client has to be
+#: willing to wait minutes, not the tens of seconds a normal API call gets —
+#: the client's own default (30s) would abort a perfectly healthy generation.
+#: ``connect`` stays short: an unreachable host is not a slow generation.
+MODEL_RUN_TIMEOUT = httpx.Timeout(600.0, connect=10.0)
+
+#: Route for a model run. NOT an ``operationId`` from ``spec/openapi.yaml`` —
+#: the vendored v2 contract declares no model routes, so this binding is
+#: hand-written and is deliberately absent from ``comfy_low.OPERATION_IDS``
+#: (the spec-coverage test asserts that set equals the spec's, exactly).
+#: Everything about the wire shape is confined to this constant and
+#: :func:`model_run_request` so it is one place to reconcile when the route is
+#: vendored into the spec and the models are regenerated from it.
+_MODEL_RUN_PATH = "/models/run"
+
 _DEFAULT_PORTS = {"http": 80, "https": 443}
 
 
+def model_run_request(
+    model: str,
+    arguments: Mapping[str, Any],
+    idempotency_key: str | None,
+) -> tuple[str, dict[str, Any], dict[str, str]]:
+    """Sans-IO ``(path, json_body, headers)`` for one model run.
+
+    The model id travels in the *body*, not the path: ids are commonly
+    provider-namespaced and contain ``/`` (``vendor/family/variant``), which in
+    a path segment needs percent-encoding that intermediaries normalize
+    inconsistently. A named body field also leaves room for sibling fields
+    later without moving the route.
+
+    ``arguments`` is copied into a plain dict so any ``Mapping`` is accepted and
+    the caller's object is never handed to the JSON encoder directly.
+    """
+    body: dict[str, Any] = {"model": model, "arguments": dict(arguments)}
+    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
+    return _MODEL_RUN_PATH, body, headers
+
+
 def _build_user_agent(client_info: str | None) -> str:
     """SDK identity sent on every request. This is request metadata (not
     telemetry — no phone-home), so adoption is measurable server-side from
@@ -490,6 +534,31 @@ def get_job_workflow(self, job_id_or_url: str, *, timeout: Any = _UNSET) -> JobW
         resp = self.raw_request("GET", path, timeout=timeout)
         return JobWorkflowResponse.model_validate(self._p.parse_or_raise(resp, (200,)))
 
+    # -- models -----------------------------------------------------------
+    def post_model_run(
+        self,
+        model: str,
+        arguments: Mapping[str, Any],
+        *,
+        idempotency_key: str | None = None,
+        timeout: Any = MODEL_RUN_TIMEOUT,
+    ) -> dict[str, Any]:
+        """POST /api/v2/models/run — run a model, awaited server-side.
+
+        One request, one response: the server does not answer until the
+        generation is complete, so the decoded body *is* the finished result.
+        That holds for a submit/poll provider too — the polling happens server
+        side, inside this call, which is why the default ``timeout`` is
+        :data:`MODEL_RUN_TIMEOUT` rather than the client's own.
+
+        The body is returned verbatim — the provider's native payload, with no
+        model class layered over it. This is not a spec operation; see
+        :data:`_MODEL_RUN_PATH`.
+        """
+        path, body, headers = model_run_request(model, arguments, idempotency_key)
+        resp = self.raw_request("POST", path, headers=headers, json=body, timeout=timeout)
+        return self._p.parse_or_raise(resp, (200, 201))
+
 
 class AsyncComfyLow:
     """Asynchronous protocol bindings — mirrors :class:`ComfyLow`."""
@@ -763,6 +832,20 @@ async def get_job_workflow(
         resp = await self.raw_request("GET", path, timeout=timeout)
         return JobWorkflowResponse.model_validate(self._p.parse_or_raise(resp, (200,)))
 
+    # -- models -----------------------------------------------------------
+    async def post_model_run(
+        self,
+        model: str,
+        arguments: Mapping[str, Any],
+        *,
+        idempotency_key: str | None = None,
+        timeout: Any = MODEL_RUN_TIMEOUT,
+    ) -> dict[str, Any]:
+        """Async :meth:`ComfyLow.post_model_run`."""
+        path, body, headers = model_run_request(model, arguments, idempotency_key)
+        resp = await self.raw_request("POST", path, headers=headers, json=body, timeout=timeout)
+        return self._p.parse_or_raise(resp, (200, 201))
+
 
 def _looks_like_path(s: str) -> bool:
     return s.startswith("http") or s.startswith("/")

src/comfy_sdk/models.py

@@ -12,15 +12,29 @@
 namespace plus a read-only view of that shared configuration; model operations
 are added to this object as they land, never to a parallel client.
 
+``run`` is the one model operation today, and it exists in exactly one form per
+client: ``Comfy().models.run(...)`` blocks, ``AsyncComfy().models.run(...)`` is
+awaited. **The awaitable form is the async client, not a suffixed method** —
+there is deliberately no ``run_async``, and there never will be: two names for
+one operation is a published signature that cannot be withdrawn once released.
+``tests/test_models_run.py`` asserts the suffix's absence rather than leaving it
+to convention.
+
 Callers do not import anything for this: ``from comfy_sdk import Comfy`` stays
 the only entry point, and ``client.models`` is the whole surface.
 """
 
 from __future__ import annotations
 
+from collections.abc import Mapping
+from typing import Any, cast
+
 import httpx
 
-from comfy_low.transport import AsyncComfyLow, ComfyLow
+from comfy_low.transport import MODEL_RUN_TIMEOUT, AsyncComfyLow, ComfyLow
+
+from ._core import new_idempotency_key
+from .exceptions import translating
 
 
 class _ModelsBase:
@@ -52,9 +66,71 @@ class Models(_ModelsBase):
     def __init__(self, low: ComfyLow) -> None:
         self._low = low
 
+    def run(
+        self,
+        model: str,
+        arguments: Mapping[str, Any],
+        *,
+        idempotency_key: str | None = None,
+        timeout: float | httpx.Timeout | None = MODEL_RUN_TIMEOUT,
+    ) -> dict[str, Any]:
+        """Run ``model`` with ``arguments`` and return the completed result.
+
+        One call, one result. It blocks until the generation is finished —
+        including for a provider the platform has to submit-and-poll, where the
+        polling happens server side inside this call, invisible to the caller.
+        There is no separate submit/await step and no ``run_async`` variant: the
+        awaitable form of this method is :meth:`AsyncModels.run` on
+        ``AsyncComfy``.
+
+        The return value is the provider's own payload, decoded from JSON and
+        handed back as-is — no wrapper class stands between the caller and the
+        fields the provider documented.
+
+        Because the server may legitimately hold the connection for minutes,
+        ``timeout`` defaults to :data:`~comfy_low.transport.MODEL_RUN_TIMEOUT`
+        (10 minutes) rather than the client's own default, which is sized for
+        ordinary API calls and would abort a healthy run. Pass a number of
+        seconds, an ``httpx.Timeout``, or ``None`` to wait indefinitely.
+
+        An ``Idempotency-Key`` is sent on every run; a fresh one is minted per
+        call unless ``idempotency_key`` is given, so an accidental exact resend
+        is the server's to reject rather than a second charged generation.
+        """
+        low = cast(ComfyLow, self._low)
+        with translating():
+            return low.post_model_run(
+                model,
+                arguments,
+                idempotency_key=idempotency_key or new_idempotency_key(),
+                timeout=timeout,
+            )
+
 
 class AsyncModels(_ModelsBase):
     """``client.models`` on :class:`~comfy_sdk.client.AsyncComfy` — mirrors :class:`Models`."""
 
     def __init__(self, low: AsyncComfyLow) -> None:
         self._low = low
+
+    async def run(
+        self,
+        model: str,
+        arguments: Mapping[str, Any],
+        *,
+        idempotency_key: str | None = None,
+        timeout: float | httpx.Timeout | None = MODEL_RUN_TIMEOUT,
+    ) -> dict[str, Any]:
+        """Awaitable :meth:`Models.run` — same arguments, same result shape.
+
+        This *is* the async form of ``run``: awaiting it on ``AsyncComfy`` is
+        the whole difference from the sync client. See :meth:`Models.run`.
+        """
+        low = cast(AsyncComfyLow, self._low)
+        with translating():
+            return await low.post_model_run(
+                model,
+                arguments,
+                idempotency_key=idempotency_key or new_idempotency_key(),
+                timeout=timeout,
+            )

tests/conftest.py

@@ -83,6 +83,25 @@ class ServerState:
     job_workflow_format: str = "api"
     job_workflow_not_found: bool = False
 
+    # --- POST /models/run (the awaited model run) ---
+    # The provider's native payload the run resolves to. Deliberately not a
+    # Comfy-shaped envelope: the SDK must hand it back untouched.
+    model_run_result: dict[str, Any] = field(
+        default_factory=lambda: {
+            "images": [{"url": "http://example.invalid/gen.png", "width": 1024, "height": 1024}],
+            "seed": 42,
+            "timings": {"inference": 3.5},
+        }
+    )
+    # Seconds the run holds the connection before answering — stands in for a
+    # generation the server polls internally, so a client whose timeout is too
+    # short aborts a healthy run.
+    model_run_delay: float = 0.0
+    # (status, code) answered instead of the result.
+    model_run_error: tuple[int, str] | None = None
+    # Status code for a successful run (201 exercises the created-shaped path).
+    model_run_status: int = 200
+
     # --- counters the tests assert on ---
     upload_count: int = 0
     from_hash_count: int = 0
@@ -106,6 +125,11 @@ class ServerState:
     # `None` before any request; `""` if a request arrived without one.
     last_auth_header: str | None = None
     last_user_agent: str | None = None
+    model_run_count: int = 0
+    last_model_run_body: dict[str, Any] | None = None
+    # Every Idempotency-Key seen on POST /models/run, in arrival order (`None`
+    # records a run that arrived without the header at all).
+    model_run_idempotency_keys: list[str | None] = field(default_factory=list)
 
 
 def _asset_json(asset_id: str, hash_: str, created_new: bool, size: int) -> dict:
@@ -168,6 +192,17 @@ class Handler(BaseHTTPRequestHandler):
         def log_message(self, *a: Any) -> None:
             pass
 
+        def handle_one_request(self) -> None:
+            # A test that deliberately aborts a slow request (an over-short
+            # client timeout against `model_run_delay`) closes the socket while
+            # this thread is still writing. That is the scenario under test,
+            # not a server fault — so don't dump a traceback for it. Only these
+            # two exception types are swallowed; anything else still surfaces.
+            try:
+                super().handle_one_request()
+            except (BrokenPipeError, ConnectionResetError):
+                self.close_connection = True
+
         # -- helpers --
         def _json(self, status: int, payload: dict, headers: dict | None = None) -> None:
             body = json.dumps(payload).encode()
@@ -364,6 +399,9 @@ def do_POST(self) -> None:
             if self.path == "/api/v2/jobs":
                 self._post_jobs()
                 return
+            if self.path == "/api/v2/models/run":
+                self._post_model_run()
+                return
             m = re.match(r"/api/v2/jobs/([^/]+)/cancel$", self.path)
             if m:
                 self._json(200, _job_json(m.group(1), "canceling"))
@@ -388,6 +426,19 @@ def _post_from_hash(self) -> None:
             else:
                 self._err(404, "blob_not_found", "no such blob")
 
+        def _post_model_run(self) -> None:
+            state.model_run_count += 1
+            state.last_model_run_body = json.loads(self._read_body() or b"{}")
+            state.model_run_idempotency_keys.append(self.headers.get("Idempotency-Key"))
+            if state.model_run_delay:
+                # The server holding the connection while it polls upstream.
+                time.sleep(state.model_run_delay)
+            if state.model_run_error is not None:
+                status, code = state.model_run_error
+                self._err(status, code, f"model run error {code}")
+                return
+            self._json(state.model_run_status, state.model_run_result)
+
         def _post_jobs(self) -> None:
             state.submit_count += 1
             body = json.loads(self._read_body() or b"{}")