"""fake_openai.py — run the course examples through the REAL openai SDK with no API key. The SDK talks HTTP. This module plugs an httpx MockTransport into a normal `OpenAI` client, so every request goes through the genuine SDK code (building requests, retries, parsing, streaming, errors) but the "server" is a small local script. Replies are either SCRIPTED (you say what the model returns, in order) or produced by a tiny rule-based fallback. from fake_openai import fake_client client = fake_client(script=["Hello from the fake model!"]) r = client.responses.create(model="gpt-5.6-luna", input="Hi") print(r.output_text) With a real key, replace `fake_client(...)` with `OpenAI()` and the rest of the code stays the same. Script items: "some text" -> an assistant message {"text": "...", "usage": (in_tokens, out)} -> a message with explicit token usage {"tool_call": "name", "arguments": {...}} -> the model asks your code to call a function {"tool_calls": [("name", {...}), ...]} -> several calls in one turn (parallel tool calls) {"status": 429} -> an HTTP error (429, 500, 400, 401, …) {"delay": 2.0} -> the server takes this long (to show timeouts) Instead of a script you can pass responder=fn: fn(request_body) returns one of the items above. That keeps replies tied to the request, which matters when requests run concurrently. """ from __future__ import annotations import asyncio import hashlib import itertools import json import math import re import time import httpx from openai import AsyncOpenAI, OpenAI _ids = itertools.count(1) def _usage(inp: int, out: int) -> dict: return {"input_tokens": inp, "output_tokens": out, "total_tokens": inp + out, "input_tokens_details": {"cached_tokens": 0}, "output_tokens_details": {"reasoning_tokens": 0}} def _approx_tokens(text: str) -> int: return max(1, round(len(text) / 4)) # the usual rule of thumb: ~4 characters per token def _message(text: str) -> dict: return {"type": "message", "id": f"msg_{next(_ids)}", "status": "completed", "role": "assistant", "content": [{"type": "output_text", "text": text, "annotations": []}]} def _function_call(name: str, arguments: dict) -> dict: n = next(_ids) return {"type": "function_call", "id": f"fc_{n}", "call_id": f"call_{n}", "name": name, "arguments": json.dumps(arguments), "status": "completed"} def _response(body: dict, output: list, usage: dict) -> dict: return {"id": f"resp_{next(_ids)}", "object": "response", "created_at": int(time.time()), "status": "completed", "model": body.get("model", "fake-model"), "output": output, "usage": usage, "parallel_tool_calls": True, "tool_choice": body.get("tool_choice", "auto"), "tools": body.get("tools", []), "instructions": body.get("instructions"), "previous_response_id": body.get("previous_response_id"), "error": None, "incomplete_details": None, "metadata": {}, "temperature": 1.0, "top_p": 1.0, "text": {"format": {"type": "text"}}} def _input_text(body: dict) -> str: inp = body.get("input", "") if isinstance(inp, str): return inp parts = [] for item in inp: content = item.get("content") if isinstance(item, dict) else None if isinstance(content, str): parts.append(content) elif isinstance(content, list): parts += [c.get("text", "") for c in content if isinstance(c, dict)] elif isinstance(item, dict) and item.get("type") == "function_call_output": parts.append(str(item.get("output", ""))) return "\n".join(parts) def embed(text: str, dims: int = 256) -> list[float]: """Deterministic toy embedding: hashed bag of words + character trigrams, unit length. Texts sharing words get similar vectors, so similarity search behaves sensibly in demos.""" vec = [0.0] * dims words = re.findall(r"[a-z0-9]+", text.lower()) features = words + [w[i:i + 3] for w in words for i in range(max(1, len(w) - 2))] for f in features: h = int(hashlib.md5(f.encode()).hexdigest(), 16) vec[h % dims] += 1.0 if (h >> 8) % 2 else -1.0 norm = math.sqrt(sum(v * v for v in vec)) or 1.0 return [v / norm for v in vec] class FakeServer: def __init__(self, script=None, responder=None): self.script = list(script or []) self.responder = responder self.requests: list[dict] = [] # every request body the SDK sent, for inspection # ------------------------------------------------------------------ routing def __call__(self, request: httpx.Request) -> httpx.Response: body = json.loads(request.content or b"{}") if request.content else {} self.requests.append({"path": request.url.path, "body": body}) path = request.url.path if path.endswith("/embeddings"): return self._embeddings(body) if path.endswith("/moderations"): return self._moderations(body) if path.endswith("/responses") or path.endswith("/chat/completions"): if self.script: item = self.script.pop(0) else: item = self.responder(body) if self.responder else None if isinstance(item, dict) and "delay" in item: read_timeout = (request.extensions.get("timeout") or {}).get("read") if read_timeout is not None and item["delay"] > read_timeout: time.sleep(read_timeout) # the client gives up after its timeout raise httpx.ReadTimeout("fake server too slow", request=request) time.sleep(item["delay"]) item = self.script.pop(0) if self.script else None if isinstance(item, dict) and "status" in item: return httpx.Response(item["status"], json={"error": { "message": item.get("message", f"fake HTTP {item['status']}"), "type": "fake_error", "code": item.get("code")}}, headers={"x-request-id": f"req_fake_{next(_ids)}", "retry-after-ms": "10"}) if path.endswith("/chat/completions"): return self._chat(body, item) return self._responses(body, item) return httpx.Response(404, json={"error": {"message": f"fake server: no route {path}"}}) # ------------------------------------------------------------------ endpoints def _responses(self, body: dict, item) -> httpx.Response: prompt = _input_text(body) + (body.get("instructions") or "") if item is None: item = f"(fake model) You said: {_input_text(body)[:80]}" if isinstance(item, str): output, text = [_message(item)], item elif "tool_call" in item: output, text = [_function_call(item["tool_call"], item.get("arguments", {}))], "" elif "tool_calls" in item: output, text = [_function_call(n, a) for n, a in item["tool_calls"]], "" else: text = item["text"] output = [_message(text)] inp, out = (item.get("usage") if isinstance(item, dict) and "usage" in item else (_approx_tokens(prompt), _approx_tokens(text or "{}"))) resp = _response(body, output, _usage(inp, out)) if body.get("stream"): return self._sse(resp, text) return httpx.Response(200, json=resp, headers={"x-request-id": f"req_fake_{next(_ids)}"}) def _sse(self, resp: dict, text: str) -> httpx.Response: seq = itertools.count() events = [("response.created", {"response": {**resp, "status": "in_progress", "output": []}})] if text: msg = resp["output"][0] events.append(("response.output_item.added", {"output_index": 0, "item": {**msg, "status": "in_progress", "content": []}})) events.append(("response.content_part.added", {"item_id": msg["id"], "output_index": 0, "content_index": 0, "part": {"type": "output_text", "text": "", "annotations": []}})) for chunk in re.findall(r"\S+\s*", text): # one "token" per word, like a real stream events.append(("response.output_text.delta", {"item_id": msg["id"], "output_index": 0, "content_index": 0, "delta": chunk, "logprobs": []})) events.append(("response.output_text.done", {"item_id": msg["id"], "output_index": 0, "content_index": 0, "text": text, "logprobs": []})) events.append(("response.content_part.done", {"item_id": msg["id"], "output_index": 0, "content_index": 0, "part": {"type": "output_text", "text": text, "annotations": []}})) events.append(("response.output_item.done", {"output_index": 0, "item": msg})) events.append(("response.completed", {"response": resp})) payload = "".join( f"event: {name}\ndata: {json.dumps({'type': name, 'sequence_number': next(seq), **data})}\n\n" for name, data in events) return httpx.Response(200, content=payload.encode(), headers={"content-type": "text/event-stream"}) def _chat(self, body: dict, item) -> httpx.Response: text = item if isinstance(item, str) else (item or {}).get("text", "(fake model) ok") prompt = " ".join(str(m.get("content", "")) for m in body.get("messages", [])) return httpx.Response(200, json={ "id": f"chatcmpl_{next(_ids)}", "object": "chat.completion", "created": int(time.time()), "model": body.get("model", "fake-model"), "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": text}}], "usage": {"prompt_tokens": _approx_tokens(prompt), "completion_tokens": _approx_tokens(text), "total_tokens": _approx_tokens(prompt) + _approx_tokens(text)}}) def _embeddings(self, body: dict) -> httpx.Response: inputs = body["input"] if isinstance(body["input"], list) else [body["input"]] dims = body.get("dimensions") or 256 data = [{"object": "embedding", "index": i, "embedding": embed(t, dims)} for i, t in enumerate(inputs)] tokens = sum(_approx_tokens(t) for t in inputs) return httpx.Response(200, json={"object": "list", "data": data, "model": body.get("model"), "usage": {"prompt_tokens": tokens, "total_tokens": tokens}}) def _moderations(self, body: dict) -> httpx.Response: inputs = body["input"] if isinstance(body["input"], list) else [body["input"]] results = [] for text in inputs: t = str(text).lower() violence = any(w in t for w in ("kill", "attack", "bomb")) harassment = any(w in t for w in ("idiot", "stupid", "hate you")) results.append({"flagged": violence or harassment, "categories": {"violence": violence, "harassment": harassment}, "category_scores": {"violence": 0.91 if violence else 0.01, "harassment": 0.88 if harassment else 0.01}}) return httpx.Response(200, json={"id": f"modr_{next(_ids)}", "model": body.get("model"), "results": results}) def fake_client(script=None, responder=None, **client_kwargs) -> OpenAI: """An OpenAI client whose HTTP traffic goes to a FakeServer. The server is at client.fake.""" server = FakeServer(script, responder) client = OpenAI(api_key="sk-fake", base_url="https://fake.openai.local/v1", http_client=httpx.Client(transport=httpx.MockTransport(server)), **client_kwargs) client.fake = server return client def fake_async_client(script=None, latency: float = 0.0, responder=None, **client_kwargs) -> AsyncOpenAI: """AsyncOpenAI against the same fake server. `latency` makes every request take that long (without blocking the event loop), and client.fake.peak records the most requests in flight.""" server = FakeServer(script, responder) server.active = server.peak = 0 async def handler(request: httpx.Request) -> httpx.Response: server.active += 1 server.peak = max(server.peak, server.active) try: if latency: await asyncio.sleep(latency) return server(request) finally: server.active -= 1 client = AsyncOpenAI(api_key="sk-fake", base_url="https://fake.openai.local/v1", http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), **client_kwargs) client.fake = server return client