Building a phone-call-style AI voice assistant in pure Python

iPhone (Shortcut: record audio + POST)
        │  multipart/form-data: audio.wav + convo_id
        ▼
┌────────────────────────────────────────────┐
│  Mac server :8765 (http.server, stdlib)    │
│                                            │
│   1. Whisper STT  (whisper.cpp / ggml-base)│
│            │ text                          │
│            ▼                               │
│   2. LLM router  (NIM → Gemini → Claude)   │
│            │ reply text                    │
│            ▼                               │
│   3. Edge TTS    (zh-TW-HsiaoChenNeural)   │
│            │ wav bytes                     │
└────────────┼───────────────────────────────┘
             ▼
    audio/wav response → iPhone plays

import subprocess, tempfile, os

def transcribe(wav_bytes: bytes) -> str:
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        f.write(wav_bytes)
        path = f.name
    try:
        subprocess.run(
            [
                "whisper-cli",
                "-m", "models/ggml-base.bin",
                "-l", "zh",
                "-f", path,
                "-otxt", "-of", path,
                "--no-prints",
            ],
            capture_output=True, text=True, timeout=30,
        )
        with open(path + ".txt") as t:
            return t.read().strip()
    finally:
        for p in (path, path + ".txt"):
            if os.path.exists(p): os.unlink(p)

import os, json, urllib.request

PROVIDERS = [
    ("nim",     "https://integrate.api.nvidia.com/v1/chat/completions",
                "meta/llama-4-maverick-17b-128e-instruct",
                os.environ["NIM_API_KEY"]),
    ("gemini",  "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
                "gemini-2.5-pro",
                os.environ["GEMINI_API_KEY"]),
    ("claude",  "https://api.anthropic.com/v1/messages",
                "claude-haiku-4-5",
                os.environ["ANTHROPIC_API_KEY"]),
]

def llm(messages: list[dict]) -> str:
    for name, url, model, key in PROVIDERS:
        try:
            if name == "claude":
                return _call_anthropic(url, key, model, messages)
            body = json.dumps({"model": model, "messages": messages,
                               "temperature": 0.6, "max_tokens": 400}).encode()
            req = urllib.request.Request(url, data=body, headers={
                "Authorization": f"Bearer {key}",
                "Content-Type": "application/json",
            })
            with urllib.request.urlopen(req, timeout=8) as r:
                data = json.loads(r.read())
                return data["choices"][0]["message"]["content"].strip()
        except Exception as e:
            print(f"[llm] {name} failed: {e}")
            continue
    return "（系統忙線中、請稍後再說一次）"

import edge_tts, asyncio
from http.server import BaseHTTPRequestHandler, HTTPServer
import cgi, io

HISTORY: dict[str, list[dict]] = {}
SYSTEM = {"role": "system",
          "content": "你是一個說繁體中文的語音助理、回答簡短、口語、自然、不超過 80 字。"}

async def synth(text: str) -> bytes:
    buf = io.BytesIO()
    comm = edge_tts.Communicate(text, voice="zh-TW-HsiaoChenNeural")
    async for chunk in comm.stream():
        if chunk["type"] == "audio":
            buf.write(chunk["data"])
    return buf.getvalue()

class Voice(BaseHTTPRequestHandler):
    def do_POST(self):
        form = cgi.FieldStorage(fp=self.rfile, headers=self.headers,
                                environ={"REQUEST_METHOD": "POST"})
        convo_id = form.getvalue("convo_id", "default")
        wav = form["audio"].file.read()

        user_text = transcribe(wav)
        hist = HISTORY.setdefault(convo_id, [SYSTEM])
        hist.append({"role": "user", "content": user_text})

        reply = llm(hist)
        hist.append({"role": "assistant", "content": reply})
        HISTORY[convo_id] = hist[-21:]  # cap at 10 turns + system

        audio = asyncio.run(synth(reply))
        self.send_response(200)
        self.send_header("Content-Type", "audio/mpeg")
        self.end_headers()
        self.wfile.write(audio)

HTTPServer(("0.0.0.0", 8765), Voice).serve_forever()