Implement support for stream responses

Using the new functionality introduced in storey 1.11.8 / mlrun/storey#605.

[ML-11876](https://iguazio.atlassian.net/browse/ML-11876)

Author: gtopper

dockerfiles/gpu/locked-requirements.txt

@@ -41,7 +41,7 @@ ENV UV_SYSTEM_PYTHON=true UV_LINK_MODE=copy UV_COMPILE_BYTECODE=1
 RUN --mount=from=uv-image,source=/uv,target=/bin/uv \
     --mount=type=cache,id=pip-${MLRUN_PYTHON_VERSION},target=/root/.cache/uv \
     --mount=type=bind,source=dockerfiles/mlrun-api/locked-requirements.txt,target=locked-requirements.txt \
-    uv pip sync  --require-hashes locked-requirements.txt --python-version ${MLRUN_PYTHON_VERSION}
+    uv pip sync locked-requirements.txt --python-version ${MLRUN_PYTHON_VERSION}
 
 WORKDIR /tmp/mlrun
 COPY *.txt *.md *.py *.toml ./

dockerfiles/jupyter/locked-requirements.txt

@@ -70,7 +70,7 @@ ENV UV_SYSTEM_PYTHON=true UV_LINK_MODE=copy UV_COMPILE_BYTECODE=1
 RUN --mount=from=uv-image,source=/uv,target=/bin/uv \
     --mount=type=cache,id=pip-${MLRUN_PYTHON_VERSION},target=/root/.cache/uv \
     --mount=type=bind,source=dockerfiles/mlrun/locked-requirements.txt,target=locked-requirements.txt \
-    uv pip install --no-deps --require-hashes -r locked-requirements.txt --python-version ${MLRUN_PYTHON_VERSION}
+    uv pip install --no-deps -r locked-requirements.txt --python-version ${MLRUN_PYTHON_VERSION}
 
 WORKDIR /tmp/mlrun
 

dockerfiles/mlrun-api/Dockerfile

@@ -97,6 +97,7 @@ class ServingSpec(nuclio_function.NuclioSpec):
         "default_class",
         "secret_sources",
         "track_models",
+        "streaming",
     ]
 
     def __init__(
@@ -154,6 +155,7 @@ def __init__(
         model_endpoint_creation_task_name=None,
         serving_spec=None,
         auth=None,
+        streaming: Optional[bool] = None,
     ):
         super().__init__(
             command=command,
@@ -212,6 +214,7 @@ def __init__(
         self.secret_sources = secret_sources or []
         self.default_content_type = default_content_type
         self.model_endpoint_creation_task_name = model_endpoint_creation_task_name
+        self.streaming = streaming
 
     @property
     def graph(self) -> Union[RouterStep, RootFlowStep]:
@@ -384,6 +387,63 @@ def set_tracking(
         if stream_args:
             self.spec.parameters["stream_args"] = stream_args
 
+    def set_streaming(self, enabled: bool = True) -> None:
+        """Enable or disable streaming mode for the serving function.
+
+        When streaming is enabled, the function handler yields results as they
+        arrive from streaming steps in the graph, allowing for real-time
+        streaming responses (e.g., for LLM token streaming).
+
+        Streaming is only supported with HTTP triggers. When streaming is enabled,
+        non-HTTP triggers cannot be added to the function.
+
+        :param enabled: Enable or disable streaming mode. Default is True.
+
+        Example::
+
+            # Create a serving function with streaming enabled
+            serving_fn = mlrun.code_to_function(kind="serving")
+            serving_fn.set_topology("flow", engine="async")
+            serving_fn.set_streaming(enabled=True)
+
+        """
+        # Validate that only HTTP triggers are configured when enabling streaming
+        if enabled:
+            # Triggers are stored as "spec.triggers.<name>" keys in the config dict
+            for key, trigger_spec in self.spec.config.items():
+                if key.startswith("spec.triggers."):
+                    trigger_name = key.split(".")[-1]
+                    trigger_kind = trigger_spec.get("kind", "http")
+                    if trigger_kind != "http":
+                        raise mlrun.errors.MLRunInvalidArgumentError(
+                            f"Streaming is only supported with HTTP triggers. "
+                            f"Found non-HTTP trigger '{trigger_name}' of kind '{trigger_kind}'. "
+                            f"Remove non-HTTP triggers before enabling streaming."
+                        )
+
+        self.spec.streaming = enabled
+
+    def add_trigger(self, name, spec):
+        """Add a nuclio trigger object/dict.
+
+        Overrides parent to validate streaming compatibility.
+
+        :param name: trigger name
+        :param spec: trigger object or dict
+        """
+        # Validate streaming compatibility
+        if self.spec.streaming:
+            trigger_spec = spec.to_dict() if hasattr(spec, "to_dict") else spec
+            trigger_kind = trigger_spec.get("kind", "http")
+            if trigger_kind != "http":
+                raise mlrun.errors.MLRunInvalidArgumentError(
+                    f"Cannot add non-HTTP trigger '{name}' (kind='{trigger_kind}') "
+                    f"when streaming is enabled. Streaming only supports HTTP triggers. "
+                    f"Either disable streaming with set_streaming(False) or use HTTP triggers only."
+                )
+
+        return super().add_trigger(name, spec)
+
     def add_model(
         self,
         key: str,
@@ -889,6 +949,13 @@ def to_job(
                 f"Cannot convert function '{self.metadata.name}' to a job because it has child functions"
             )
 
+        if self.spec.streaming:
+            raise mlrun.errors.MLRunInvalidArgumentError(
+                f"Cannot convert function '{self.metadata.name}' to a job because streaming "
+                f"is enabled. Streaming functions return real-time HTTP responses and cannot "
+                f"run as batch jobs. Please disable streaming with set_streaming(False) first."
+            )
+
         self._add_steps_requirements()
 
         spec = pod_runtime.KubeResourceSpec(