/
Watashicuvu
/
agentic-tools
Обзор
Документация
Войти
/
Watashicuvu
/
agentic-tools
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/localscript_api/sse.py
38 строк
1 KB
Якуб
mts update
12 апр 2026, 21:19
12 апр 2026, 21:19
bff6dcd
Код
Авторство
О чём код?
"""SSE streaming helpers для LocalScript API. Использует sse-starlette для Server-Sent Events. """ from __future__ import annotations import json from typing import AsyncGenerator from sse_starlette.sse import ServerSentEvent def format_sse_event(event_type: str, data: str, meta: dict | None = None) -> ServerSentEvent: """Создаёт SSE событие в стандартном формате. Args: event_type: Тип события (question, answer, validating, code, error) data: Данные события meta: Дополнительные метаданные Returns: ServerSentEvent для отправки клиенту """ payload = {"type": event_type, "data": data} if meta: payload["meta"] = meta return ServerSentEvent(data=json.dumps(payload, ensure_ascii=False), event=event_type) async def sse_error(error_message: str) -> AsyncGenerator[ServerSentEvent, None]: """SSE поток с одной ошибкой.""" yield format_sse_event("error", error_message) async def sse_code_only(code: str) -> AsyncGenerator[ServerSentEvent, None]: """SSE поток с одним событием code.""" yield format_sse_event("code", code)