Add streaming support for /v1/responses compatibility

app/routes.py

@@ -6,6 +6,7 @@ import re
 import threading
 import time
 import urllib.request
+import urllib.parse
 from uuid import uuid4
 
 from fastapi import APIRouter, HTTPException, Request
@@ -196,6 +197,22 @@ def _call_local_chat_completions(chat_body: dict, auth_header: str, x_api_key: s
         return 500, {"error": str(exc)}
 
 
+def _open_local_chat_completions_stream(chat_body: dict, auth_header: str, x_api_key: str):
+    local_port = os.getenv("PORT", "7860")
+    headers = {"Content-Type": "application/json", "Accept": "text/event-stream"}
+    if auth_header:
+        headers["Authorization"] = auth_header
+    if x_api_key:
+        headers["x-api-key"] = x_api_key
+    req = urllib.request.Request(
+        f"http://127.0.0.1:{local_port}/v1/chat/completions",
+        data=json.dumps(chat_body).encode("utf-8"),
+        headers=headers,
+        method="POST",
+    )
+    return urllib.request.urlopen(req, timeout=180)
+
+
 def _chat_completion_to_response_payload(chat_payload: dict) -> dict:
     choice = ((chat_payload.get("choices") or [{}])[0])
     message = choice.get("message") or {}
@@ -245,16 +262,176 @@ def _chat_completion_to_response_payload(chat_payload: dict) -> dict:
     }
 
 
+def _sse_event(data: dict) -> str:
+    return f"event: {data['type']}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
+
+
+def _response_base(response_id: str, model: str) -> dict:
+    return {
+        "id": response_id,
+        "object": "response",
+        "created_at": int(time.time()),
+        "status": "in_progress",
+        "error": None,
+        "incomplete_details": None,
+        "instructions": None,
+        "model": model,
+        "output": [],
+        "parallel_tool_calls": True,
+        "tools": [],
+        "top_p": 1,
+        "temperature": 1,
+        "text": {"format": {"type": "text"}},
+        "metadata": {},
+        "usage": None,
+    }
+
+
+def _stream_responses_from_chat(chat_body: dict, auth_header: str, x_api_key: str, model: str):
+    response_id = f"resp_{uuid4().hex}"
+    message_id = f"msg_{uuid4().hex}"
+    seq = 1
+
+    yield _sse_event({"type": "response.created", "response": _response_base(response_id, model), "sequence_number": seq})
+    seq += 1
+    yield _sse_event({"type": "response.in_progress", "response": _response_base(response_id, model), "sequence_number": seq})
+    seq += 1
+    yield _sse_event(
+        {
+            "type": "response.output_item.added",
+            "output_index": 0,
+            "item": {
+                "id": message_id,
+                "type": "message",
+                "status": "in_progress",
+                "role": "assistant",
+                "content": [],
+            },
+            "sequence_number": seq,
+        }
+    )
+    seq += 1
+    yield _sse_event(
+        {
+            "type": "response.content_part.added",
+            "item_id": message_id,
+            "output_index": 0,
+            "content_index": 0,
+            "part": {"type": "output_text", "text": "", "annotations": []},
+            "sequence_number": seq,
+        }
+    )
+    seq += 1
+
+    full_text = []
+    usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
+
+    try:
+        with _open_local_chat_completions_stream(chat_body, auth_header, x_api_key) as resp:
+            for raw_line in resp:
+                line = raw_line.decode("utf-8", errors="replace").strip()
+                if not line or not line.startswith("data: "):
+                    continue
+                payload = line[6:]
+                if payload == "[DONE]":
+                    break
+                try:
+                    chunk = json.loads(payload)
+                except json.JSONDecodeError:
+                    continue
+                choice = ((chunk.get("choices") or [{}])[0])
+                delta = choice.get("delta") or {}
+                text_delta = delta.get("content")
+                if text_delta:
+                    full_text.append(text_delta)
+                    yield _sse_event(
+                        {
+                            "type": "response.output_text.delta",
+                            "item_id": message_id,
+                            "output_index": 0,
+                            "content_index": 0,
+                            "delta": text_delta,
+                            "sequence_number": seq,
+                        }
+                    )
+                    seq += 1
+                if "usage" in chunk and isinstance(chunk["usage"], dict):
+                    usage = {
+                        "input_tokens": chunk["usage"].get("prompt_tokens", 0),
+                        "output_tokens": chunk["usage"].get("completion_tokens", 0),
+                        "total_tokens": chunk["usage"].get("total_tokens", 0),
+                    }
+    except urllib.error.HTTPError as exc:
+        raw = exc.read().decode("utf-8", errors="replace")
+        message = raw
+        try:
+            parsed = json.loads(raw)
+            message = parsed.get("error") or parsed.get("detail") or raw
+        except Exception:
+            pass
+        yield _sse_event({"type": "error", "error": {"message": str(message)}, "sequence_number": seq})
+        return
+    except Exception as exc:
+        yield _sse_event({"type": "error", "error": {"message": str(exc)}, "sequence_number": seq})
+        return
+
+    final_text = "".join(full_text)
+    yield _sse_event(
+        {
+            "type": "response.output_text.done",
+            "item_id": message_id,
+            "output_index": 0,
+            "content_index": 0,
+            "text": final_text,
+            "sequence_number": seq,
+        }
+    )
+    seq += 1
+    yield _sse_event(
+        {
+            "type": "response.content_part.done",
+            "item_id": message_id,
+            "output_index": 0,
+            "content_index": 0,
+            "part": {"type": "output_text", "text": final_text, "annotations": []},
+            "sequence_number": seq,
+        }
+    )
+    seq += 1
+    yield _sse_event(
+        {
+            "type": "response.output_item.done",
+            "output_index": 0,
+            "item": {
+                "id": message_id,
+                "type": "message",
+                "status": "completed",
+                "role": "assistant",
+                "content": [{"type": "output_text", "text": final_text, "annotations": []}],
+            },
+            "sequence_number": seq,
+        }
+    )
+    seq += 1
+    completed = _response_base(response_id, model)
+    completed["status"] = "completed"
+    completed["output"] = [
+        {
+            "id": message_id,
+            "type": "message",
+            "status": "completed",
+            "role": "assistant",
+            "content": [{"type": "output_text", "text": final_text, "annotations": []}],
+        }
+    ]
+    completed["usage"] = usage
+    yield _sse_event({"type": "response.completed", "response": completed, "sequence_number": seq})
+
+
 @router.post("/v1/responses")
 async def responses_api(request: Request):
     try:
         req_data = await request.json()
-        if bool(req_data.get("stream", False)):
-            return JSONResponse(
-                status_code=400,
-                content={"error": "/v1/responses streaming is not supported yet. Use /v1/chat/completions with stream=true."},
-            )
-
         model = req_data.get("model")
         messages = _responses_input_to_messages(req_data)
         if not model or not messages:
@@ -266,7 +443,7 @@ async def responses_api(request: Request):
         chat_body = {
             "model": model,
             "messages": messages,
-            "stream": False,
+            "stream": bool(req_data.get("stream", False)),
         }
         if "tools" in req_data:
             chat_body["tools"] = _responses_tools_to_chat_tools(req_data.get("tools"))
@@ -277,6 +454,17 @@ async def responses_api(request: Request):
         if "top_p" in req_data:
             chat_body["top_p"] = req_data.get("top_p")
 
+        if bool(req_data.get("stream", False)):
+            return StreamingResponse(
+                _stream_responses_from_chat(
+                    chat_body,
+                    request.headers.get("Authorization", ""),
+                    request.headers.get("x-api-key", ""),
+                    model,
+                ),
+                media_type="text/event-stream",
+            )
+
         status_code, chat_payload = await __import__("asyncio").to_thread(
             _call_local_chat_completions,
             chat_body,