From 69bef239457cea780819d4ccda7f3418115f530a Mon Sep 17 00:00:00 2001
From: Ethan Wong <ewong@anl.gov>
Date: Fri, 24 Jul 2026 14:55:46 -0500
Subject: [PATCH] responses: allow streaming on `DirectAPI`-backed endpoints

Enables OpenAI Responses API and Anthropic Messages API streaming for
`DirectAPI`-backed endpoints like on Metis and Minerva. This primarily
benefits coding agents which require streaming via these APIs for
3rd-party endpoint utilization.
---
 resource_server_async/services.py        | 20 +++++++++++++++++---
 resource_server_async/views/anthropic.py |  8 --------
 resource_server_async/views/openai.py    |  6 ------
 3 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/resource_server_async/services.py b/resource_server_async/services.py
index 2d36fb90..76372a71 100644
--- a/resource_server_async/services.py
+++ b/resource_server_async/services.py
@@ -22,7 +22,7 @@
 )
 
 from .clusters import BaseCluster
-from .endpoints import BaseEndpoint
+from .endpoints import BaseEndpoint, DirectAPIEndpoint
 from .errors import (
     BatchOngoing,
     BatchUnavailable,
@@ -243,6 +243,8 @@ async def submit_openai_inference_request(
     framework: str,
     payload: OpenAIRequestPayload,
 ) -> StreamingHttpResponse | Any:
+    is_openai_responses = isinstance(payload, OpenAIResponsesPydantic)
+    is_anthropic_messages = isinstance(payload, AnthropicMessagesPydantic)
     if isinstance(payload, OpenAIChatCompletionsPydantic):
         stream = payload.stream or False
         prompt = payload.model_dump(include={"messages"})["messages"]
@@ -252,10 +254,10 @@ async def submit_openai_inference_request(
     elif isinstance(payload, OpenAIEmbeddingsPydantic):
         stream = False
         prompt = payload.input
-    elif isinstance(payload, OpenAIResponsesPydantic):
+    elif is_openai_responses:
         stream = payload.stream or False
         prompt = payload.model_dump(include={"input"}, mode="json")["input"]
-    elif isinstance(payload, AnthropicMessagesPydantic):
+    elif is_anthropic_messages:
         stream = payload.stream or False
         prompt = payload.model_dump(include={"messages"}, mode="json")["messages"]
     else:
@@ -288,6 +290,18 @@ async def submit_openai_inference_request(
         f"endpoint_slug: {endpoint.endpoint_slug} - user: {context.user.username}"
     )
 
+    if (
+        stream
+        and (is_openai_responses or is_anthropic_messages)
+        and not isinstance(endpoint, DirectAPIEndpoint)
+    ):
+        # We don't support streaming on non-DirectAPI backed endpoints currently
+        raise UnsupportedEndpoint(
+            "Streaming is not supported for the "
+            f"{'OpenAI Responses' if is_openai_responses else 'Anthropic Messages'}"
+            " API on this endpoint. Re-issue this request with 'stream': false."
+        )
+
     # Overwrite model name in case it has been updated via endpoint slug mapping
     payload.model = endpoint.model
 
diff --git a/resource_server_async/views/anthropic.py b/resource_server_async/views/anthropic.py
index bc1b3d74..f700378c 100644
--- a/resource_server_async/views/anthropic.py
+++ b/resource_server_async/views/anthropic.py
@@ -3,7 +3,6 @@
 
 from ninja import Router
 
-from ..errors import UnsupportedEndpoint
 from ..logging import get_request_context
 from ..schemas.anthropic_messages import AnthropicMessagesPydantic
 from ..schemas.auth import AuthedRequest
@@ -22,13 +21,6 @@ async def create_message(
     framework: str,
     payload: AnthropicMessagesPydantic,
 ) -> Any:
-
-    if payload.stream:
-        raise UnsupportedEndpoint(
-            "Streaming is not supported for the Anthropic Messages API on "
-            "this gateway. Re-issue the request with 'stream': false."
-        )
-
     return await submit_openai_inference_request(
         get_request_context(), cluster_name, framework, payload
     )
diff --git a/resource_server_async/views/openai.py b/resource_server_async/views/openai.py
index 8f93223f..44d2a408 100644
--- a/resource_server_async/views/openai.py
+++ b/resource_server_async/views/openai.py
@@ -3,7 +3,6 @@
 
 from ninja import Router
 
-from ..errors import UnsupportedEndpoint
 from ..logging import get_request_context
 from ..schemas.auth import AuthedRequest
 from ..schemas.openai_chat_completions import (
@@ -65,11 +64,6 @@ async def create_response(
     framework: str,
     payload: OpenAIResponsesPydantic,
 ) -> Any:
-    if payload.stream:
-        raise UnsupportedEndpoint(
-            "Streaming is not supported for the OpenAI Responses API on this "
-            "gateway. Re-issue the request with 'stream': false."
-        )
     return await submit_openai_inference_request(
         get_request_context(), cluster_name, framework, payload
     )