INNER CODE UNIT · Python

sse_vector_analytics

marc-shade/world-intel-mcp · src/world_intel_mcp/dashboard/app.py:597

async def sse_vector_analytics(request):
    """SSE endpoint — pushes vector store analytics every 30 seconds."""

    async def event_generator():
        while True:
            try:
                payload = await _fetch_vector_analytics()
                yield f"data: {json.dumps(payload, default=str)}\n\n"
            except asyncio.CancelledError:
                return
            except Exception as exc:
                logger.exception("Vector analytics SSE tick failed")
                yield f"data: {json.dumps({'error': str(exc)})}\n\n"
            await asyncio.sleep(30)

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

View source record →