Build a synthetic inference response for ``/responses`` or ``/chat/completions``. Dispatches by URL and the request body's ``stream`` flag: ``/responses`` streams an SSE event sequence (or returns a buffered Responses object when ``stream`` is false), ``/chat/completions`` streams chat-
(request: httpx.Request, text: str = SYNTHETIC_TEXT)
| 161 | |
| 162 | |
| 163 | def build_inference_response(request: httpx.Request, text: str = SYNTHETIC_TEXT) -> httpx.Response: |
| 164 | """Build a synthetic inference response for ``/responses`` or ``/chat/completions``. |
| 165 | |
| 166 | Dispatches by URL and the request body's ``stream`` flag: ``/responses`` |
| 167 | streams an SSE event sequence (or returns a buffered Responses object when |
| 168 | ``stream`` is false), ``/chat/completions`` streams chat-completion chunks |
| 169 | (or returns a buffered completion). |
| 170 | """ |
| 171 | body = request.content # already drained when send_request is called |
| 172 | wants_stream = _wants_stream(body) |
| 173 | url = str(request.url).lower() |
| 174 | |
| 175 | if "/responses" in url: |
| 176 | if not wants_stream: |
| 177 | return httpx.Response( |
| 178 | 200, |
| 179 | headers={"content-type": "application/json"}, |
| 180 | content=json.dumps(responses_events(text)[-1]["response"]).encode(), |
| 181 | ) |
| 182 | stream_body = "".join(sse(e["type"], e) for e in responses_events(text)) |
| 183 | return httpx.Response( |
| 184 | 200, |
| 185 | headers={"content-type": "text/event-stream"}, |
| 186 | content=stream_body.encode(), |
| 187 | ) |
| 188 | |
| 189 | if "/chat/completions" in url and wants_stream: |
| 190 | base = { |
| 191 | "id": "chatcmpl-stub-1", |
| 192 | "object": "chat.completion.chunk", |
| 193 | "created": 1, |
| 194 | "model": "claude-sonnet-4.5", |
| 195 | } |
| 196 | chunks = [ |
| 197 | { |
| 198 | **base, |
| 199 | "choices": [ |
| 200 | { |
| 201 | "index": 0, |
| 202 | "delta": {"role": "assistant", "content": ""}, |
| 203 | "finish_reason": None, |
| 204 | } |
| 205 | ], |
| 206 | }, |
| 207 | { |
| 208 | **base, |
| 209 | "choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}], |
| 210 | }, |
| 211 | { |
| 212 | **base, |
| 213 | "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], |
| 214 | "usage": {"prompt_tokens": 5, "completion_tokens": 7, "total_tokens": 12}, |
| 215 | }, |
| 216 | ] |
| 217 | stream_body = ( |
| 218 | "".join("data: " + json.dumps(c) + "\n\n" for c in chunks) + "data: [DONE]\n\n" |
| 219 | ) |
| 220 | return httpx.Response( |
searching dependent graphs…