phase 0.4: SSE streaming end-to-end (chat-chunk wire path + OpenAI-format /v1)

publisher's chat_handler can now yield chunks (sync iterator or async
iterator). hub forwards as SSE in OpenAI streaming format. external
clients use openai library's stream=True natively.

protocol additions:
- protocol.py: chat_chunk(delta, request_id, done, finish_reason) — new
  envelope type. publisher emits one or more, terminates with done=True.

server side:
- server.py /v1/chat/completions: when body.stream==true, returns
  text/event-stream StreamingResponse. Hub holds a queue per request_id;
  publisher's chat-chunk envelopes flow into the queue, the SSE generator
  consumes and emits OpenAI-format `data: {...}\n\n` lines plus the final
  `data: [DONE]\n\n` sentinel.
- server.py proxy_chat: gains stream=True path that hands back a queue
  + request_id instead of a future.
- server.py ws_publish: chat-response from a non-streaming publisher
  paired with a streaming queue caller is auto-wrapped into a single
  chunk + done — keeps backwards compatibility for handlers that don't
  yet support streaming.

client side:
- client.py _handle_chat: detects async iterators (yields chunks via
  chat-chunk), sync iterators when stream was requested, and falls back
  to the existing single-shot string/dict shape. operator's chat handler
  can be a yield-generator now.

example handler shape:
    def my_handler(messages, options):
        for word in "the quick brown fox".split():
            yield word + " "

invoking from outside:
    import openai
    client = openai.OpenAI(base_url=..., api_key=...)
    for chunk in client.chat.completions.create(
        model="my-ai", messages=[...], stream=True,
    ):
        print(chunk.choices[0].delta.content, end="", flush=True)

works through hub natively.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Zawwarsami16

zhub/client.py

@@ -23,7 +23,7 @@
 
 from .manifest import Capability, Manifest, chat_only_manifest
 from .protocol import (
-    Envelope, register_publisher, register_connection, chat_request,
+    Envelope, register_publisher, register_connection, chat_request, chat_chunk,
     invoke_request, invoke_result,
 )
 from .errors import AuthError, ConnectionError as ZhubConnectionError
@@ -184,10 +184,27 @@ async def runner() -> None:
 async def _handle_chat(pub: ZhubPublication, ws, env: Envelope) -> None:
     messages = env.payload.get("messages", [])
     options = {k: v for k, v in env.payload.items() if k != "messages"}
+    streaming_requested = bool(options.get("stream"))
     try:
         result = pub.chat_handler(messages, options)
+
+        # Coroutines
         if asyncio.iscoroutine(result):
             result = await result
+
+        # Sync iterators / async iterators — streaming
+        if hasattr(result, "__aiter__"):
+            async for chunk in result:
+                await ws.send(chat_chunk(str(chunk), env.request_id).to_json())
+            await ws.send(chat_chunk("", env.request_id, done=True, finish_reason="stop").to_json())
+            return
+        if streaming_requested and hasattr(result, "__iter__") and not isinstance(result, (str, dict, bytes)):
+            for chunk in result:
+                await ws.send(chat_chunk(str(chunk), env.request_id).to_json())
+            await ws.send(chat_chunk("", env.request_id, done=True, finish_reason="stop").to_json())
+            return
+
+        # Single-shot
         if isinstance(result, str):
             payload = {"text": result, "finish_reason": "stop"}
         elif isinstance(result, dict):

zhub/protocol.py

@@ -116,6 +116,25 @@ def chat_response(text: str, request_id: str, finish_reason: str = "stop",
     )
 
 
+def chat_chunk(delta: str, request_id: str, done: bool = False,
+               finish_reason: Optional[str] = None) -> Envelope:
+    """Streaming chunk — incremental delta of a chat response.
+
+    The publisher sends one or more chat-chunk envelopes followed by a final
+    chunk with done=True. The hub forwards them to the HTTP client as SSE
+    events in OpenAI streaming format.
+    """
+    return Envelope(
+        type="chat-chunk",
+        request_id=request_id,
+        payload={
+            "delta": delta,
+            "done": done,
+            "finish_reason": finish_reason,
+        },
+    )
+
+
 def invoke_request(connection_id: str, capability: str,
                    args: dict[str, Any]) -> Envelope:
     return Envelope(

zhub/server.py

@@ -19,7 +19,7 @@
 
 import argparse
 import asyncio
-import json
+import json  # noqa: F401  -- used in inline SSE serialization
 import logging
 import secrets
 import time
@@ -29,7 +29,7 @@
 
 try:
     from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
-    from fastapi.responses import JSONResponse
+    from fastapi.responses import JSONResponse, StreamingResponse
     import uvicorn
 except ImportError as e:
     raise SystemExit(
@@ -152,13 +152,21 @@ async def unregister_connection(self, ai_name: str, connection_id: str) -> None:
 
     async def proxy_chat(self, ai_name: str, messages: list[dict[str, Any]],
                          model: str, temperature: float, max_tokens: int,
-                         timeout: float = 60.0) -> dict[str, Any]:
-        """Route an HTTP chat request to the publisher and await its response."""
+                         timeout: float = 60.0,
+                         stream: bool = False) -> dict[str, Any]:
+        """Route an HTTP chat request to the publisher and await its response.
+        If stream=True the future delivers a queue of streaming chunks instead."""
         publisher = self.publishers.get(ai_name)
         if publisher is None:
             raise LookupError("publisher not registered")
         env = chat_request(messages=messages, model=model,
-                           temperature=temperature, max_tokens=max_tokens)
+                           temperature=temperature, max_tokens=max_tokens,
+                           extras={"stream": True} if stream else None)
+        if stream:
+            queue: asyncio.Queue = asyncio.Queue()
+            publisher.pending[env.request_id] = queue  # type: ignore[assignment]
+            await publisher.websocket.send_text(env.to_json())
+            return {"_stream_queue": queue, "_request_id": env.request_id}
         future: asyncio.Future = asyncio.get_running_loop().create_future()
         publisher.pending[env.request_id] = future
         try:
@@ -243,7 +251,7 @@ async def manifest(ai_name: str) -> JSONResponse:
         return JSONResponse(m)
 
     @app.post("/{ai_name}/v1/chat/completions")
-    async def chat_completions(ai_name: str, request: Request) -> JSONResponse:
+    async def chat_completions(ai_name: str, request: Request):
         body = await request.json()
         api_key_header = request.headers.get("authorization", "").removeprefix("Bearer ").strip()
         if hub.lookup_by_api_key(api_key_header) != ai_name:
@@ -253,14 +261,72 @@ async def chat_completions(ai_name: str, request: Request) -> JSONResponse:
         model = body.get("model", "default")
         temperature = float(body.get("temperature", 0.4))
         max_tokens = int(body.get("max_tokens", 4096))
+        stream = bool(body.get("stream", False))
+
+        if stream:
+            try:
+                response = await hub.proxy_chat(
+                    ai_name, messages, model, temperature, max_tokens, stream=True,
+                )
+            except LookupError:
+                raise HTTPException(404, "AI offline")
+
+            queue: asyncio.Queue = response["_stream_queue"]
+            request_id = response["_request_id"]
+
+            async def event_stream():
+                created = int(time.time())
+                completion_id = "chatcmpl-" + new_request_id()[:16]
+                while True:
+                    chunk = await queue.get()
+                    if chunk is None:
+                        break
+                    delta_text = chunk.get("delta", "")
+                    done = chunk.get("done", False)
+                    finish_reason = chunk.get("finish_reason")
+                    if done:
+                        # final chunk per OpenAI streaming spec
+                        sse = {
+                            "id": completion_id,
+                            "object": "chat.completion.chunk",
+                            "created": created,
+                            "model": model,
+                            "choices": [{
+                                "index": 0,
+                                "delta": {},
+                                "finish_reason": finish_reason or "stop",
+                            }],
+                        }
+                        yield f"data: {json.dumps(sse)}\n\n"
+                        yield "data: [DONE]\n\n"
+                        break
+                    sse = {
+                        "id": completion_id,
+                        "object": "chat.completion.chunk",
+                        "created": created,
+                        "model": model,
+                        "choices": [{
+                            "index": 0,
+                            "delta": {"role": "assistant", "content": delta_text},
+                            "finish_reason": None,
+                        }],
+                    }
+                    yield f"data: {json.dumps(sse)}\n\n"
+
+            return StreamingResponse(
+                event_stream(),
+                media_type="text/event-stream",
+                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+            )
+
         try:
             response = await hub.proxy_chat(ai_name, messages, model, temperature, max_tokens)
         except LookupError:
             raise HTTPException(404, "AI offline")
         except asyncio.TimeoutError:
             raise HTTPException(504, "AI did not respond in time")
 
-        # Wrap into OpenAI-style response shape
+        # Wrap into OpenAI-style response shape (non-streaming)
         text = response.get("text", "")
         usage = response.get("usage", {})
         return JSONResponse({
@@ -298,7 +364,25 @@ async def ws_publish(websocket: WebSocket) -> None:
                 elif env.type == "chat-response" and ai_name:
                     publisher = hub.publishers.get(ai_name)
                     if publisher and env.request_id in publisher.pending:
-                        publisher.pending[env.request_id].set_result(env.payload)
+                        target = publisher.pending[env.request_id]
+                        if isinstance(target, asyncio.Queue):
+                            # streaming caller — convert non-streaming response to single chunk
+                            await target.put({"delta": env.payload.get("text", ""),
+                                              "done": False})
+                            await target.put({"done": True,
+                                              "finish_reason": env.payload.get("finish_reason", "stop")})
+                            await target.put(None)
+                        else:
+                            target.set_result(env.payload)
+
+                elif env.type == "chat-chunk" and ai_name:
+                    publisher = hub.publishers.get(ai_name)
+                    if publisher and env.request_id in publisher.pending:
+                        target = publisher.pending[env.request_id]
+                        if isinstance(target, asyncio.Queue):
+                            await target.put(env.payload)
+                            if env.payload.get("done"):
+                                await target.put(None)
 
                 elif env.type == "invoke-request" and ai_name:
                     # Publisher wants to call a connected client.