feat(vllm-omni): accept application/json on video routes

Restore JSON->multipart conversion in the SageMaker routing middleware for
the form-data video routes (/v1/videos, /v1/videos/sync). These upstream
handlers only accept multipart/form-data, so a SageMaker caller sending
application/json with route=/v1/videos/sync got a 400. The middleware now
converts a JSON object body into multipart/form-data when the resolved route
is one of FORM_DATA_ROUTES, so JSON callers work without changing the proven
multipart path (byte-for-byte passthrough).

Stays on vllm-omni 0.21.0rc1 (no framework bump): the 0.22.0 release ships a
TrackingArgumentParser that crashes on argparse append actions (our
--middleware flag), so the version bump is deferred until the upstream fix is
released. Bumps dlc_minor_version 3 -> 4 for the middleware change.

Author: Jyothirmaikottu

.github/config/image/vllm-omni/ec2-amzn2023.yml

@@ -25,7 +25,7 @@ build:
   torch_cuda_arch_list: "7.5;8.0;8.6;8.9;9.0;10.0;11.0;12.0+PTX"
   install_kv_connectors: "true"
   dlc_major_version: "1"
-  dlc_minor_version: "3"
+  dlc_minor_version: "4"
   use_sccache: "true"
 
 release:

.github/config/image/vllm-omni/sagemaker-amzn2023.yml

@@ -26,7 +26,7 @@ build:
   torch_cuda_arch_list: "7.5;8.0;8.6;8.9;9.0;10.0;11.0;12.0+PTX"
   install_kv_connectors: "true"
   dlc_major_version: "1"
-  dlc_minor_version: "3"
+  dlc_minor_version: "4"
   use_sccache: "true"
 
 release:

.github/config/model-tests/vllm-omni-model-tests.yml

@@ -104,6 +104,21 @@ smoke-test:
       test_request: 'prompt=a dog running on a beach&num_frames=17&num_inference_steps=4&size=480x320&seed=42'
       validate: "binary_size_gt:1000"
 
+    # Sync route via application/json: the SageMaker routing middleware
+    # (omni_sagemaker_serve.py) converts the JSON object into a
+    # multipart/form-data body before handing it to the upstream
+    # /v1/videos/sync handler, which only accepts form data. Exercises the
+    # JSON content-type path so SageMaker callers can send JSON instead of
+    # building a multipart body by hand.
+    - name: "wan2.1-t2v-1.3b-sync-json"
+      s3_model: "wan2.1-t2v-1.3b.tar.gz"
+      fleet: "x86-g6exl-runner"
+      extra_args: ""
+      route: "/v1/videos/sync"
+      content_type: "application/json"
+      test_request: '{"prompt": "a dog running on a beach", "num_frames": 17, "num_inference_steps": 4, "size": "480x320", "seed": 42}'
+      validate: "binary_size_gt:1000"
+
     # Wan2.1-VACE: unified video creation/editing pipeline (WanVACEPipeline,
     # added in vllm-omni #1885). Distinct from WanPipeline T2V — accepts
     # text + optional video/mask/reference image. 1.3B variant fits L40S.

.github/workflows/vllm-omni.pr.yml

@@ -1,6 +1,7 @@
 name: "PR - vLLM-Omni"
 
 on:
+  workflow_dispatch:
   pull_request:
     branches: [main]
     types: [opened, reopened, synchronize]

scripts/docker/vllm/omni_sagemaker_serve.py

@@ -8,11 +8,15 @@
 If no route is specified, falls through to vLLM's built-in /invocations
 handler (chat/completion/embed).
 
-Clients targeting routes that require multipart/form-data (e.g. /v1/videos)
-must send the request with ContentType="multipart/form-data; boundary=..."
-and a pre-built multipart body. SageMaker InvokeEndpoint accepts arbitrary
-ContentType values and forwards them to the model server unchanged, so no
-in-middleware conversion is needed.
+Some routes (e.g. /v1/videos, /v1/videos/sync) are implemented upstream as
+multipart/form-data handlers. Clients may target them either with a
+pre-built multipart/form-data body, or — for convenience over SageMaker
+InvokeEndpoint — with a plain application/json object. When the resolved
+route is one of FORM_DATA_ROUTES and the request arrives as application/json,
+this middleware transparently converts the JSON object into a
+multipart/form-data body (each top-level key becomes a form field) so the
+upstream handler receives the encoding it expects. multipart/form-data bodies
+are passed through unchanged.
 
 Usage: vllm serve --omni --middleware omni_sagemaker_serve.SageMakerRouteMiddleware
 
@@ -25,13 +29,19 @@
 no-op and SageMaker /invocations will return 404 for non-default routes.
 """
 
+import json
 import logging
 import re
+import uuid
 
-from starlette.types import ASGIApp, Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 logger = logging.getLogger("omni_sagemaker")
 
+# Routes whose upstream handlers expect multipart/form-data. A JSON body
+# targeting one of these is converted to multipart before it reaches vLLM.
+FORM_DATA_ROUTES = frozenset({"/v1/videos", "/v1/videos/sync"})
+
 
 def _parse_route(headers: list[tuple[bytes, bytes]]) -> str | None:
     """Extract route=<path> from SageMaker custom attributes header."""
@@ -42,11 +52,79 @@ def _parse_route(headers: list[tuple[bytes, bytes]]) -> str | None:
     return None
 
 
+def _content_type(headers: list[tuple[bytes, bytes]]) -> str:
+    """Return the bare content-type (no parameters), lowercased."""
+    for key, value in headers:
+        if key.lower() == b"content-type":
+            return value.decode().split(";", 1)[0].strip().lower()
+    return ""
+
+
+def _field_value(value: object) -> str:
+    """Render a non-null JSON value as a multipart form field string.
+
+    Scalars map to their natural string form; bool maps to "true"/"false";
+    nested objects/arrays are re-encoded as JSON. None is handled by the
+    caller (_build_multipart drops null keys) and never reaches here.
+    """
+    if isinstance(value, str):
+        return value
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, (int, float)):
+        return str(value)
+    return json.dumps(value)
+
+
+def _escape_field_name(name: str) -> str:
+    """Escape a form-field name for the Content-Disposition header.
+
+    RFC 7578 requires CR/LF and double-quote to be escaped inside name="...".
+    Without this a key containing " or a newline would corrupt the header.
+    """
+    return name.replace("\\", "\\\\").replace('"', '\\"').replace("\r", "%0D").replace("\n", "%0A")
+
+
+def _build_multipart(fields: dict, boundary: str) -> bytes:
+    # JSON null has no faithful form-data representation: a form field is
+    # always a present string. Emitting the literal "null" would make the
+    # upstream handler see a present-but-bogus value instead of an absent one,
+    # bypassing its default-on-missing logic. Dropping the key reproduces what
+    # a caller omitting the field would send, which is the closest match.
+    parts = [
+        f'--{boundary}\r\nContent-Disposition: form-data; name="{_escape_field_name(k)}"'
+        f"\r\n\r\n{_field_value(v)}\r\n"
+        for k, v in fields.items()
+        if v is not None
+    ]
+    parts.append(f"--{boundary}--\r\n")
+    return "".join(parts).encode()
+
+
+def _replay(body: bytes) -> Receive:
+    """Build a receive callable that yields `body` as a single http.request."""
+    sent = False
+
+    async def receive() -> Message:
+        nonlocal sent
+        if not sent:
+            sent = True
+            return {"type": "http.request", "body": body, "more_body": False}
+        return {"type": "http.disconnect"}
+
+    return receive
+
+
 class SageMakerRouteMiddleware:
     """ASGI middleware that reroutes /invocations based on CustomAttributes.
 
     Explicit route via header -> rewrites path to that endpoint.
     No route specified -> falls through to vLLM's built-in /invocations handler.
+
+    When the resolved route expects multipart/form-data (FORM_DATA_ROUTES) but
+    the body arrives as application/json, the JSON object is converted to a
+    multipart/form-data body so the upstream handler receives its expected
+    encoding.
     """
 
     def __init__(self, app: ASGIApp) -> None:
@@ -61,4 +139,50 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
                 scope["path"] = route
                 scope["raw_path"] = route.encode()
 
+                if (
+                    route in FORM_DATA_ROUTES
+                    and _content_type(scope.get("headers", [])) == "application/json"
+                ):
+                    scope, receive = await self._json_to_multipart(scope, receive)
+
         await self.app(scope, receive, send)
+
+    async def _json_to_multipart(self, scope: Scope, receive: Receive) -> tuple[Scope, Receive]:
+        """Drain the JSON body, convert it to multipart, and return a new
+        (scope, receive) carrying the rewritten content-type and body.
+
+        On any failure (non-object JSON, parse error) the original body is
+        replayed untouched so the upstream handler produces its normal error
+        response rather than this middleware masking it.
+        """
+        body = b""
+        more_body = True
+        while more_body:
+            message = await receive()
+            body += message.get("body", b"")
+            more_body = message.get("more_body", False)
+
+        try:
+            parsed = json.loads(body)
+            if not isinstance(parsed, dict):
+                raise ValueError("JSON body for a form route must be an object")
+        except (ValueError, json.JSONDecodeError) as exc:
+            logger.warning("Could not convert JSON body to multipart: %s", exc)
+            return scope, _replay(body)
+
+        boundary = uuid.uuid4().hex
+        new_body = _build_multipart(parsed, boundary)
+        content_type = f"multipart/form-data; boundary={boundary}".encode()
+
+        headers = [
+            (k, v)
+            for k, v in scope["headers"]
+            if k.lower() not in (b"content-type", b"content-length")
+        ]
+        headers.append((b"content-type", content_type))
+        headers.append((b"content-length", str(len(new_body)).encode()))
+        scope = dict(scope)
+        scope["headers"] = headers
+
+        logger.info("Converted application/json body to multipart/form-data")
+        return scope, _replay(new_body)