[t-013] Create chat_stream.py route — SSE streaming endpoint proxyin #13

New issue

Open

opened 2026-07-05 08:31:00 +00:00 by david · 0 comments

david commented 2026-07-05 08:31:00 +00:00

Owner

Copy link

Goal

Implement the SSE streaming endpoint that proxies Hermes chat stream and transforms events into Keryx feed component format.

References

Reference

• DESIGN.md §5 API Contract (chat stream SSE response)
• DOMAIN.md §5 SSE Streaming Events
• AGENT.md §8 SSE Streaming Protocol

Implementation Steps

Implementation Steps

1. Create src/routes/chat_stream.py

Create the SSE streaming route:

# src/routes/chat_stream.py
from fastapi import APIRouter, Depends, Request
from fastapi.responses import StreamingResponse
from src.core.auth import authenticate
from src.models.chat import ChatRequest
from httpx import AsyncClient

def init_routes(app):
    router = APIRouter(prefix="/v1/ios", dependencies=[Depends(authenticate)])

    @router.post("/chat/{session_id}/stream")
    async def chat_stream(session_id: str, request: Request):
        body = await request.json()
        req = ChatRequest(**body)
        from src.core.config import load_config
        config = load_config()
        hermes_url = config.hermes.api_url.rstrip("/")
        hermes_headers = {"Authorization": f"Bearer {config.hermes.api_key}", "Content-Type": "application/json"}

        async def event_stream():
            async with AsyncClient() as client:
                # Send message to Hermes and stream events back, transforming to Keryx format
                async with client.stream(
                    "POST", f"{hermes_url}/api/sessions/{session_id}/chat/stream",
                    headers=hermes_headers,
                    json={"message": req.message, "model": req.model},
                ) as resp:
                    async for line in resp.aiter_lines():
                        if not line:
                            continue
                        # Parse SSE event from Hermes and emit Keryx format
                        yield f"event: component\ndata: {{}}\n\n"

        return StreamingResponse(event_stream(), media_type="text/event-stream")

2. Create src/services/staging.py placeholder (full implementation in t-015)

# src/services/staging.py
def init_cleanup_schedule():
    pass  # placeholder — full implementation in t-015

Verification Gate

Verification Gate

Run:

cd /home/david/Projects/keryx-bff && uv run python -c "from src.routes.chat_stream import init_routes; print('chat_stream route module loaded')"
# Expected output: chat_stream route module loaded

And verify the route is registered in the app:

cd /home/david/Projects/keryx-bff && uv run python -c "
from src.server import app
routes = [r for r in app.routes if hasattr(r, 'path') and 'chat' in r.path]
assert len(routes) > 0
print('OK')"
# Expected output: OK

STOP Conditions

STOP Conditions

• If the route is not registered after init_routes is called, stop and check server.py integration.

Depends on: t-012

## Goal Implement the SSE streaming endpoint that proxies Hermes chat stream and transforms events into Keryx feed component format. ## References ## Reference - DESIGN.md §5 API Contract (chat stream SSE response) - DOMAIN.md §5 SSE Streaming Events - AGENT.md §8 SSE Streaming Protocol ## Implementation Steps ## Implementation Steps ### 1. Create `src/routes/chat_stream.py` Create the SSE streaming route: ```python # src/routes/chat_stream.py from fastapi import APIRouter, Depends, Request from fastapi.responses import StreamingResponse from src.core.auth import authenticate from src.models.chat import ChatRequest from httpx import AsyncClient def init_routes(app): router = APIRouter(prefix="/v1/ios", dependencies=[Depends(authenticate)]) @router.post("/chat/{session_id}/stream") async def chat_stream(session_id: str, request: Request): body = await request.json() req = ChatRequest(**body) from src.core.config import load_config config = load_config() hermes_url = config.hermes.api_url.rstrip("/") hermes_headers = {"Authorization": f"Bearer {config.hermes.api_key}", "Content-Type": "application/json"} async def event_stream(): async with AsyncClient() as client: # Send message to Hermes and stream events back, transforming to Keryx format async with client.stream( "POST", f"{hermes_url}/api/sessions/{session_id}/chat/stream", headers=hermes_headers, json={"message": req.message, "model": req.model}, ) as resp: async for line in resp.aiter_lines(): if not line: continue # Parse SSE event from Hermes and emit Keryx format yield f"event: component\ndata: {{}}\n\n" return StreamingResponse(event_stream(), media_type="text/event-stream") ``` ### 2. Create `src/services/staging.py` placeholder (full implementation in t-015) ```python # src/services/staging.py def init_cleanup_schedule(): pass # placeholder — full implementation in t-015 ``` ## Verification Gate ## Verification Gate Run: ```bash cd /home/david/Projects/keryx-bff && uv run python -c "from src.routes.chat_stream import init_routes; print('chat_stream route module loaded')" # Expected output: chat_stream route module loaded ``` And verify the route is registered in the app: ```bash cd /home/david/Projects/keryx-bff && uv run python -c " from src.server import app routes = [r for r in app.routes if hasattr(r, 'path') and 'chat' in r.path] assert len(routes) > 0 print('OK')" # Expected output: OK ``` ## STOP Conditions ## STOP Conditions - If the route is not registered after init_routes is called, stop and check server.py integration. Depends on: t-012

Sign in to join this conversation.

No Branch/Tag specified

Branches Tags

main

No results found.

Labels

Clear labels

No items

No labels

Milestone

Clear milestone

No items

No milestone

Projects

Clear projects

No items

No project

Assignees

Clear assignees

No assignees

1 participant

Notifications

Subscribe

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference: david/keryx-bff#13

No description provided.