(text: str, resp_id: str)
| 80 | |
| 81 | |
| 82 | def _sse_body(text: str, resp_id: str) -> bytes: |
| 83 | out = "".join( |
| 84 | f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" |
| 85 | for event in responses_events(text, resp_id) |
| 86 | ) |
| 87 | return out.encode("utf-8") |
| 88 | |
| 89 | |
| 90 | async def _start_fake_upstream() -> _Upstream: |
no test coverage detected
searching dependent graphs…