chatbot1-2  Exercise 3: A Bot on an OpenAI-Compatible Server, End to End
==========================================================================

This uses the scripted local server from the chapter (tests/support.py, shown in full below), so no key and no real model are involved.

ex3.py (save in the project root):

  import asyncio
  from bot.core import Bot
  from bot.providers.openai_compat import OpenAICompatProvider
  from tests.support import ScriptedServer

  def ok(text):
      return {"model": "local-test", "choices": [{"message": {"role": "assistant", "content": text}, "finish_reason": "stop"}],
              "usage": {"prompt_tokens": 5, "completion_tokens": 3}}

  async def main():
      with ScriptedServer() as srv:
          srv.respond(200, ok("Hello Ada!"))
          srv.respond(200, ok("Your name is Ada."))
          provider = OpenAICompatProvider(srv.url, "local-test")          # no key, like a local server
          bot = Bot(provider, system="Be brief.")
          print(await bot.handle("s", "My name is Ada"))
          print(await bot.handle("s", "What is my name?"))
          for path, headers, body in srv.requests:
              print(path, "| authorization header sent:", "authorization" in headers,
                    "| messages on the wire:", [(m["role"]) for m in body["messages"]])
          await provider.aclose()
  asyncio.run(main())

Run with:  py ex3.py

Output:

  Hello Ada!
  Your name is Ada.
  /chat/completions | authorization header sent: False | messages on the wire: ['system', 'user']
  /chat/completions | authorization header sent: False | messages on the wire: ['system', 'user', 'assistant', 'user']

tests/support.py (needed by the script above and by the tests):

  import json
  import socketserver
  import threading
  import time
  from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


  class QuickServer(ThreadingHTTPServer):
      def server_bind(self):
          # HTTPServer.server_bind calls socket.getfqdn(), which can take ~1s on Windows
          socketserver.TCPServer.server_bind(self)
          self.server_port = self.server_address[1]


  class ScriptedServer:
      """A tiny local HTTP server that answers with a queue of canned responses.

      Each queued item is (status, headers, body, delay_seconds). Every request received is
      recorded in .requests as (path, headers, parsed JSON body).
      """

      def __init__(self) -> None:
          self.queue: list[tuple[int, dict, object, float]] = []
          self.requests: list[tuple[str, dict, dict]] = []
          outer = self

          class Handler(BaseHTTPRequestHandler):
              def do_POST(self):
                  length = int(self.headers.get("content-length", 0))
                  body = json.loads(self.rfile.read(length) or b"{}")
                  outer.requests.append((self.path, {k.lower(): v for k, v in self.headers.items()}, body))
                  status, headers, payload, delay = outer.queue.pop(0)
                  if delay:
                      time.sleep(delay)
                  data = json.dumps(payload).encode() if not isinstance(payload, bytes) else payload
                  try:
                      self.send_response(status)
                      self.send_header("content-type", "application/json")
                      self.send_header("content-length", str(len(data)))
                      for k, v in headers.items():
                          self.send_header(k, v)
                      self.end_headers()
                      self.wfile.write(data)
                  except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
                      pass

              def log_message(self, *args):
                  pass

          self._server = QuickServer(("127.0.0.1", 0), Handler)
          self.port = self._server.server_address[1]
          self.url = f"http://127.0.0.1:{self.port}"
          self._thread = threading.Thread(target=lambda: self._server.serve_forever(poll_interval=0.01), daemon=True)

      def __enter__(self):
          self._thread.start()
          return self

      def __exit__(self, *exc):
          self._server.shutdown()
          self._server.server_close()

      def respond(self, status=200, payload=None, headers=None, delay=0.0):
          self.queue.append((status, headers or {}, payload if payload is not None else {}, delay))

WHY THIS WORKS AS AN ANSWER
---------------------------
The Bot and its history code are exactly the ones from chapter 1; only the provider changed, and the provider talked real HTTP to a real (local) server. The output shows the three things this chapter
promised. The system prompt reached the wire as a first message with role system, which is how the Chat Completions style carries it (the Claude provider would put it in a top-level field instead).
No Authorization header was sent, because the provider was built with no key, which is correct for a local server. And the second request carried the whole conversation (system, user, assistant, user), so the
memory from chapter 1 works unchanged through a different provider. The replies were scripted by the test server, so this shows plumbing, not model behaviour. Pointing the same code at a real Ollama server is the
subject of chapter 3 and needs a local runtime that this machine does not have.