diff --git a/src/backend/base/langflow/api/router.py b/src/backend/base/langflow/api/router.py
index f49c3fe3cb..e86a0867d2 100644
--- a/src/backend/base/langflow/api/router.py
+++ b/src/backend/base/langflow/api/router.py
@@ -1,5 +1,6 @@
 # Router for base api
 from fastapi import APIRouter
+from lfx.schema.workflow import WORKFLOW_EXECUTION_RESPONSES
 from lfx.services.settings.feature_flags import FEATURE_FLAGS
 from lfx.workflow.host import WorkflowHost
 from lfx.workflow.router import create_workflow_router
@@ -138,7 +139,12 @@ router_v2.include_router(registration_router_v2)
 _workflow_host = LangflowWorkflowHost()
 assert isinstance(_workflow_host, WorkflowHost)  # noqa: S101
 router_v2.include_router(
-    create_workflow_router(_workflow_host, developer_api_guard=False, auto_register_job_routes=False)
+    create_workflow_router(
+        _workflow_host,
+        developer_api_guard=False,
+        auto_register_job_routes=False,
+        responses=WORKFLOW_EXECUTION_RESPONSES,
+    )
 )
 # The langflow-owned durable routes: GET status, POST /stop, GET /{job_id}/events.
 router_v2.include_router(workflow_background_router_v2)
diff --git a/src/backend/tests/unit/api/v2/test_workflow_agui.py b/src/backend/tests/unit/api/v2/test_workflow_agui.py
index 1ccdd274bd..7452e5ac52 100644
--- a/src/backend/tests/unit/api/v2/test_workflow_agui.py
+++ b/src/backend/tests/unit/api/v2/test_workflow_agui.py
@@ -340,20 +340,22 @@ class TestV2WorkflowAdmission:
     async def test_denial_echoes_requested_identifier_not_resolved_uuid(self, monkeypatch: pytest.MonkeyPatch):
         """A denial on a flow referenced by endpoint name must not leak the resolved UUID."""
         from langflow.api.v2 import workflow as workflow_module
+        from langflow.api.v2.workflow_host import LangflowWorkflowHost
         from lfx.workflow.actions import WorkflowAction
+        from lfx.workflow.host import ResolvedFlow
 
         resolved_uuid = uuid4()
         flow = SimpleNamespace(id=resolved_uuid, user_id=uuid4(), workspace_id=None, folder_id=None)
+        resolved = ResolvedFlow(flow_id="my-endpoint", graph=flow)
 
         async def _deny(*_args, **_kwargs):
             raise HTTPException(status_code=403, detail="denied")
 
         monkeypatch.setattr(workflow_module, "ensure_flow_permission", _deny)
 
+        # Through the host, so the requested-id wiring is what's exercised, not just the helper.
         with pytest.raises(HTTPException) as exc_info:
-            await workflow_module.authorize_flow_action(
-                SimpleNamespace(id=uuid4()), flow, WorkflowAction.EXECUTE, requested_id="my-endpoint"
-            )
+            await LangflowWorkflowHost().authorize(SimpleNamespace(id=uuid4()), resolved, WorkflowAction.EXECUTE)
 
         assert exc_info.value.status_code == 404
         assert exc_info.value.detail["code"] == "FLOW_NOT_FOUND"
diff --git a/src/lfx/src/lfx/workflow/router.py b/src/lfx/src/lfx/workflow/router.py
index d0edfbde75..a777256f11 100644
--- a/src/lfx/src/lfx/workflow/router.py
+++ b/src/lfx/src/lfx/workflow/router.py
@@ -1,15 +1,17 @@
 """The shared v2 workflow HTTP router.
 
-lfx owns the entire env-neutral handler body: request parsing, stream-protocol
-validation, sync/stream dispatch, the single SSE framing loop, error -> HTTP
-mapping, and the ``developer_api_enabled`` router guard (which reads lfx's own
-settings service). The host (:mod:`lfx.workflow.host`) supplies only the
-DB/tenant-bound capabilities — caller resolution, fetch-and-authorize, the
-request session, and whether durable background runs exist.
+lfx owns the env-neutral handler body: request parsing, stream-protocol
+validation, sync/stream dispatch, error -> HTTP mapping, the lfx-default SSE
+framing loop, and the ``developer_api_enabled`` router guard (which reads lfx's
+own settings service). The host (:mod:`lfx.workflow.host`) supplies the
+DB/tenant-bound capabilities (caller resolution, fetch-and-authorize, the
+request session, durable background runs) and may override sync/stream
+execution; the langflow host does, keeping its richer SSE pipeline.
 
-Both runtimes (langflow backend and bare ``lfx serve``) mount the same router
-via :func:`create_workflow_router`, so neither the wire contract nor the SSE
-shape can drift by construction.
+Both runtimes mount the same router via :func:`create_workflow_router`, so the
+request/response contract and the admission + error mapping are single-sourced.
+The SSE framing is single-sourced only for hosts using the lfx default (bare
+``lfx serve``); langflow overrides it.
 """
 
 from __future__ import annotations
@@ -250,6 +252,7 @@ def create_workflow_router(
     tags: tuple[str, ...] = ("Workflow",),
     developer_api_guard: bool = True,
     auto_register_job_routes: bool = True,
+    responses: dict | None = None,
 ) -> APIRouter:
     """Build the v2 workflow router bound to ``host``.
 
@@ -284,6 +287,7 @@ def create_workflow_router(
     @router.post(
         "",
         response_model=None,
+        responses=responses or {},
         summary="Execute Workflow (v2 sync or stream)",
     )
     async def execute_workflow(request: WorkflowRunRequest, http_request: Request, background_tasks: BackgroundTasks):
diff --git a/src/lfx/tests/unit/workflow/test_router_contract.py b/src/lfx/tests/unit/workflow/test_router_contract.py
index 2fafc093fd..2ef9d008c2 100644
--- a/src/lfx/tests/unit/workflow/test_router_contract.py
+++ b/src/lfx/tests/unit/workflow/test_router_contract.py
@@ -13,7 +13,7 @@ from typing import Any
 from uuid import uuid4
 
 import pytest
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.testclient import TestClient
 from lfx.components.input_output import ChatInput, ChatOutput
 from lfx.graph import Graph
@@ -113,7 +113,25 @@ def test_authorize_receives_workflow_action_member():
     assert action is WorkflowAction.EXECUTE
 
 
+def test_unknown_protocol_422_precedes_flow_fetch():
+    """An unknown stream_protocol must 422 before get_flow runs, even if the flow is missing."""
+
+    class _NoFlowHost(_FakeHost):
+        async def get_flow(self, flow_id, caller):  # noqa: ARG002
+            raise HTTPException(status_code=404, detail="get_flow should not run for a bad protocol")
+
+    client = _client(_NoFlowHost(_echo_graph()))
+    resp = client.post(
+        "/workflows",
+        json={"flow_id": _FLOW_ID, "input_value": "hi", "mode": "stream", "stream_protocol": "nope"},
+    )
+    assert resp.status_code == 422, resp.text
+    assert resp.json()["detail"]["code"] == "UNKNOWN_STREAM_PROTOCOL"
+
+
 def test_sse_frame_shape_and_terminal_frame():
+    # Pins the lfx-default framing (id line first), which bare serve uses. The langflow
+    # host overrides stream_response, so this does not assert cross-runtime SSE identity.
     host = _FakeHost(_echo_graph())
     client = _client(host)
     resp = client.post(