This repository has no description
27 kB
610 lines
1"""hermes-voice: talk to Hermes through OpenAI's full-duplex gpt-live-1.
2
3The browser holds a WebRTC session with gpt-live-1, which listens and speaks at
4the same time and chats on its own. Whenever the user asks for real work, the
5voice model emits a client delegation; the page hands the recent transcript to
6this server, which starts it as a Hermes run on the API server (POST /v1/runs,
7one Hermes session id per conversation), follows the run's event stream and
8streams the reply back. The page then gives the answer to the voice model,
9which says it aloud.
10
11A run lives independently of any HTTP connection, so a turn survives the page
12going away. A newer delegation for the same conversation stops the previous
13run (the user corrected or replaced the request); hanging up just ends the
14call, and the run finishes whatever the user asked for.
15
16This server exists so neither secret reaches the browser: it exchanges the
17page's SDP offer for a Live session with the OpenAI key, and it calls Hermes
18with the API server's bearer token. That token is root-equivalent for the
19Hermes host, so whoever can use this server can drive the agent: every API
20route requires the page's own access token (--access-token-file), and serving
21beyond loopback without one is refused.
22
23Vendor contract: https://developers.openai.com/api/docs/guides/live-delegation
24"""
25
26import argparse
27import hmac
28import json
29import logging
30import os
31import re
32import socket
33import threading
34import time
35import urllib.error
36import urllib.request
37from http import HTTPStatus
38from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
39from pathlib import Path
40
41log = logging.getLogger("hermes-voice")
42
43OPENAI_BASE_URL = "https://api.openai.com/v1"
44DEFAULT_MODEL = "gpt-live-1"
45DEFAULT_VOICE = "marin"
46# gpt-live-1 voices as OpenAI's voice picker lists them: language, regional
47# influence, presentation and character, natural voices first, then the
48# generated ones. The docs' voice table and the openai SDK (3.28.0) list only
49# part of them; /v1/live/sessions accepts the German, French, Hindi, Japanese
50# and Korean voices as well, while an unknown name is refused with 403.
51BUILT_IN_VOICES = {
52 "gleam": "English, North American, female: warm and conversational",
53 "ripple": "English, Australian, male: warm and conversational",
54 "vesper": "English, British, male: clear and direct",
55 "willow": "English, Irish, female: expressive and bright",
56 "stone": "English, Irish, male: calm and precise",
57 "meridian": "English, North American, male: grounded and measured",
58 "bossa": "Portuguese, Brazilian, female: clear and direct",
59 "tempo": "Portuguese, Brazilian, male: grounded and measured",
60 "cedar": "English, North American, male: open and upbeat",
61 "marin": "English, North American, female: animated and earnest",
62 "nuri": "German, female: warm and animated",
63 "juni": "German, female: bright and expressive",
64 "flitz": "German, male: rich and playful",
65 "brise": "French, female: clear and conversational",
66 "sillage": "French, male: warm and grounded",
67 "nira": "Hindi, female: soft and thoughtful",
68 "harema": "Japanese, female: gentle and engaging",
69 "shida": "Japanese, male: quiet and grounded",
70 "noeul": "Korean, female: airy and expressive",
71 "quartz": "English, Australian, female, generated: warm and conversational",
72 "alloy": "English, North American, male, generated: balanced and clear",
73 "ash": "English, North American, male, generated: savvy and relaxed",
74 "ballad": "English, North American, male, generated: cheerful and candid",
75 "coral": "English, North American, female, generated: easygoing and versatile",
76 "echo": "English, North American, male, generated: bright and inquisitive",
77 "sage": "English, North American, female, generated: confident and optimistic",
78 "shimmer": "English, North American, female, generated: composed and direct",
79 "verse": "English, North American, male, generated: calm and affirming",
80 "beacon": "English, Filipino, male, generated: clear and personable",
81 "delta": "English, Southern U.S., female, generated: warm and easygoing",
82 "cinder": "English, Southern U.S., male, generated: steady and grounded",
83}
84# gpt-live-1 bills every second a session is open, silence included; the page
85# closes the session after this much inactivity.
86DEFAULT_IDLE_SECONDS = 90
87MAX_BODY = 256 * 1024
88# The run event stream sends a keepalive every 10s, so a read that stalls this
89# long means the API server is gone.
90EVENTS_READ_TIMEOUT = 60
91SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
92# Hermes run ids are "run_<hex>", approval request ids hex.
93RUN_ID_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,256}$")
94
95# Frontend persona, adapted from Hermes's own (tools/voice_live.py): role,
96# style and a labelled delegation policy. The backend carries the real
97# instructions, tools and memory.
98PERSONA = """You are Hermes, lass's warm, upbeat and genuinely helpful voice assistant. Sound friendly and glad to help: positive and encouraging, never curt, annoyed, bored or sarcastic, even when delivering bad news or when the user is short with you. Talk fast: speak at a rapid, energetic pace, like someone who talks quickly by nature, with no slow drawn-out words and no long pauses between sentences. Keep replies short; a brief warm acknowledgment before the facts is good.
99
100Backchannel policy: Use moderate backchannels. Acknowledge naturally without competing with the main response.
101
102Interruption policy: Stop speaking when the user interrupts. Listen to what they say.
103
104Delegation policy:
105Backend tools:
106- Hermes agent: a full AI agent with tools. It can run commands on lass's machines, read and edit files, manage the calendar, browse the web, search, and remember things across sessions. It is the one who actually does work and knows facts.
107
108Delegate to the backend when:
109- The user asks a question that needs facts, current information, or careful reasoning.
110- The user asks you to do, check, find, make, fix, run or remember anything.
111- A correction changes work already requested.
112
113Do not delegate to the backend when:
114- The user greets you, makes small talk, or asks you to repeat a result already provided.
115- You need a brief clarification to understand the request.
116
117Delegate before giving an answer that depends on backend work. Do not guess the result while waiting; say briefly and cheerfully that you are on it, then wait for the result.
118
119Approval policy:
120Hermes sometimes needs the user's approval before it runs a command. When you are told so, explain in plain words what Hermes wants to do and ask whether to approve or deny it. When the user answers, delegate to the backend so the answer reaches Hermes. Never decide an approval yourself."""
121
122# Ephemeral system prompt for every delegated Hermes turn (Hermes layers system
123# messages on top of its core prompt without storing them).
124BACKEND_NOTE = """This turn is a delegation from a live spoken conversation. The user message is a speech transcript: it may contain mis-hearings, phonetic spellings of technical terms, hesitations and later corrections; use the latest intent. Your reply will be spoken aloud by a friendly voice model that paraphrases it: answer in warm, plain conversational sentences, keep it short (a few sentences unless the user asked for detail), no markdown, no lists, no code blocks, no URLs read out character by character. Do the work with your tools as usual; only the final facts need to be spoken. Report an action as done only after it actually succeeded."""
125
126
127def glossary_text(glossary):
128 """Glossary entries as prompt lines: `term` or `term (sounds like "…")`."""
129 lines = []
130 for entry in glossary:
131 if isinstance(entry, str):
132 lines.append(f"- {entry}")
133 else:
134 term = entry["term"]
135 spoken = entry.get("spoken")
136 lines.append(f'- {term} (sounds like "{spoken}")' if spoken else f"- {term}")
137 return "\n".join(lines)
138
139
140class Config:
141 def __init__(self, raw, openai_key, hermes_url, hermes_key):
142 self.model = raw.get("model", DEFAULT_MODEL)
143 self.voice = raw.get("voice", DEFAULT_VOICE)
144 self.idle_seconds = int(raw.get("idle_seconds", DEFAULT_IDLE_SECONDS))
145 self.glossary = raw.get("glossary", [])
146 self.extra_instructions = raw.get("instructions", "").strip()
147 self.openai_key = openai_key
148 self.hermes_url = hermes_url.rstrip("/")
149 self.hermes_key = hermes_key
150
151 def live_instructions(self):
152 parts = [PERSONA]
153 if self.glossary:
154 parts.append(
155 "Technical terms the user says often. Understand them when spoken and "
156 "say them like this:\n" + glossary_text(self.glossary)
157 )
158 if self.extra_instructions:
159 parts.append(self.extra_instructions)
160 return "\n\n".join(parts)
161
162 def backend_note(self):
163 if not self.glossary:
164 return BACKEND_NOTE
165 return (
166 BACKEND_NOTE
167 + "\n\nTerms the transcript may contain as phonetic misspellings:\n"
168 + glossary_text(self.glossary)
169 )
170
171 def voices(self):
172 """Voices the page may pick, name -> description: the built-ins plus
173 the configured one (which may be a custom voice id)."""
174 voices = dict(BUILT_IN_VOICES)
175 voices.setdefault(self.voice, "custom voice")
176 return voices
177
178
179def voice_param(voice):
180 # Custom voices are objects ({"id": "voice_…"}); built-ins are names.
181 if voice.startswith("voice_"):
182 return {"id": voice}
183 return voice
184
185
186def live_history(lines):
187 """Seed history for a new Live session from the page's transcript lines
188 (spoken dialogue only; Hermes answers were spoken through the voice)."""
189 items = []
190 dialogue = [line for line in lines if line.get("speaker") in ("user", "assistant")]
191 for line in dialogue[-24:]:
192 text = str(line.get("text", "")).strip()[:1200]
193 if not text:
194 continue
195 if line["speaker"] == "assistant":
196 content = {"type": "output_text", "text": text}
197 else:
198 content = {"type": "input_text", "text": text}
199 items.append({"type": "message", "role": line["speaker"], "content": [content]})
200 return items
201
202
203def create_live_session(config, sdp, history, voice):
204 session = {
205 "model": config.model,
206 "instructions": config.live_instructions(),
207 "audio": {"output": {"voice": voice_param(voice)}},
208 "delegation": {"type": "client"},
209 # The page is an untrusted frontend: it may only answer delegations and
210 # hang up, not rewrite the session.
211 "client": {
212 "data_channel": {
213 "allowed_client_events": [
214 "session.commentary.append",
215 "session.thinking.append",
216 "session.close",
217 ]
218 }
219 },
220 }
221 if history:
222 session["input"] = history
223 body = json.dumps(
224 {"session": session, "transport": {"type": "webrtc", "sdp": sdp}}
225 ).encode()
226 request = urllib.request.Request(
227 f"{OPENAI_BASE_URL}/live/sessions",
228 data=body,
229 method="POST",
230 headers={
231 "Authorization": f"Bearer {config.openai_key}",
232 "Content-Type": "application/json",
233 },
234 )
235 with urllib.request.urlopen(request, timeout=30) as response:
236 answer = json.load(response)
237 return {"session_id": answer["session"]["id"], "sdp": answer["transport"]["sdp"]}
238
239
240def transcript_message(lines):
241 rendered = []
242 for line in lines:
243 text = str(line.get("text", "")).strip()
244 if text:
245 who = "Assistant" if line.get("speaker") == "assistant" else "User"
246 rendered.append(f"{who}: {text}")
247 return "Spoken conversation since the last request, newest last:\n" + "\n".join(
248 rendered
249 )
250
251
252def iter_sse(response):
253 """Yield (event, data) pairs from a text/event-stream response."""
254 event, data = None, []
255 for raw in response:
256 line = raw.decode("utf-8", "replace").rstrip("\r\n")
257 if not line:
258 if data:
259 yield event, "\n".join(data)
260 event, data = None, []
261 elif line.startswith("event:"):
262 event = line[6:].strip()
263 elif line.startswith("data:"):
264 data.append(line[5:].lstrip())
265 if data:
266 yield event, "\n".join(data)
267
268
269def hermes_request(config, method, path, payload=None, timeout=30):
270 """Open a request to the Hermes API server (the caller closes it)."""
271 request = urllib.request.Request(
272 f"{config.hermes_url}{path}",
273 data=None if payload is None else json.dumps(payload).encode(),
274 method=method,
275 headers={
276 "Authorization": f"Bearer {config.hermes_key}",
277 "Content-Type": "application/json",
278 },
279 )
280 return urllib.request.urlopen(request, timeout=timeout)
281
282
283def hermes_call(config, method, path, payload=None):
284 with hermes_request(config, method, path, payload) as response:
285 return json.load(response)
286
287
288def hermes_start_run(config, hermes_session, lines):
289 """Start one Hermes turn as a run; return its run id."""
290 # A plain string input: Hermes then loads the history of session_id itself.
291 return hermes_call(
292 config,
293 "POST",
294 "/v1/runs",
295 {
296 "input": transcript_message(lines),
297 "instructions": config.backend_note(),
298 "session_id": hermes_session,
299 },
300 )["run_id"]
301
302
303def hermes_follow_run(config, run_id):
304 """Follow a run's events; yield NDJSON-ready dicts (tool, delta, approval,
305 done, error). Leaving early only detaches: the run keeps going."""
306 with hermes_request(
307 config, "GET", f"/v1/runs/{run_id}/events", timeout=EVENTS_READ_TIMEOUT
308 ) as response:
309 for _, data in iter_sse(response):
310 payload = json.loads(data)
311 event = payload.get("event", "")
312 if event in ("tool.started", "tool.completed"):
313 status = "running" if event == "tool.started" else "completed"
314 if payload.get("error"):
315 status = "error"
316 yield {"type": "tool", "status": status, "label": payload.get("tool")}
317 elif event == "approval.request":
318 # The run blocks until POST /v1/runs/{run_id}/approval answers.
319 yield {
320 "type": "approval",
321 "run_id": run_id,
322 "request_id": payload.get("request_id"),
323 "command": payload.get("command"),
324 "description": payload.get("description"),
325 }
326 elif event == "run.completed":
327 # Only the final answer: message.delta also streams commentary.
328 yield {"type": "delta", "text": payload.get("output") or ""}
329 yield {"type": "done", "finish_reason": "stop"}
330 return
331 elif event in ("run.failed", "run.cancelled", "run.interrupted"):
332 status = event.removeprefix("run.")
333 error = payload.get("error") or f"Hermes run {status}"
334 yield {"type": "done", "finish_reason": status, "error": error}
335 return
336 yield {"type": "error", "message": "Hermes event stream ended early"}
337
338
339def hermes_approve(config, run_id, request_id, choice):
340 """Answer a pending approval of a Hermes run."""
341 answer = {"choice": choice}
342 if request_id:
343 answer["request_id"] = request_id
344 return hermes_call(config, "POST", f"/v1/runs/{run_id}/approval", answer)
345
346
347# The active run per Hermes session, so a newer delegation can stop it.
348RUNS = {}
349RUNS_LOCK = threading.Lock()
350
351
352def take_run(hermes_session, run_id=None):
353 """Remove and return the session's run (only if it is run_id, when given)."""
354 with RUNS_LOCK:
355 if run_id is None or RUNS.get(hermes_session) == run_id:
356 return RUNS.pop(hermes_session, None)
357 return None
358
359
360def stop_run(config, run_id):
361 """POST /v1/runs/{run_id}/stop; failures are only logged."""
362 try:
363 hermes_call(config, "POST", f"/v1/runs/{run_id}/stop")
364 log.info("stopped run %s", run_id)
365 except (urllib.error.URLError, OSError, ValueError) as error:
366 log.warning("stopping run %s failed: %s", run_id, error)
367
368
369class Handler(BaseHTTPRequestHandler):
370 server_version = "hermes-voice"
371 config: Config
372 static: Path
373 access_token: "str | None" = None
374
375 def authorized(self):
376 """Check the page's bearer token; answer 401 when it is wrong."""
377 if self.access_token is None:
378 return True
379 header = self.headers.get("Authorization", "")
380 scheme, _, token = header.partition(" ")
381 if scheme == "Bearer" and hmac.compare_digest(
382 token.encode(), self.access_token.encode()
383 ):
384 return True
385 self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "access token required"})
386 return False
387
388 def log_message(self, format, *args):
389 log.info("%s %s", self.address_string(), format % args)
390
391 def send_json(self, status, payload):
392 body = json.dumps(payload).encode()
393 self.send_response(status)
394 self.send_header("Content-Type", "application/json")
395 self.send_header("Content-Length", str(len(body)))
396 self.send_header("Cache-Control", "no-store")
397 self.end_headers()
398 self.wfile.write(body)
399
400 def read_json(self):
401 length = int(self.headers.get("Content-Length") or 0)
402 if length <= 0 or length > MAX_BODY:
403 raise ValueError("bad body length")
404 return json.loads(self.rfile.read(length))
405
406 def do_GET(self):
407 if self.path == "/config":
408 if not self.authorized():
409 return
410 self.send_json(
411 HTTPStatus.OK,
412 {
413 "model": self.config.model,
414 "voice": self.config.voice,
415 "voices": self.config.voices(),
416 "idle_seconds": self.config.idle_seconds,
417 },
418 )
419 return
420 names = {
421 "/": ("index.html", "text/html; charset=utf-8"),
422 "/app.js": ("app.js", "text/javascript; charset=utf-8"),
423 "/style.css": ("style.css", "text/css; charset=utf-8"),
424 }
425 if self.path not in names:
426 self.send_error(HTTPStatus.NOT_FOUND)
427 return
428 name, content_type = names[self.path]
429 body = (self.static / name).read_bytes()
430 self.send_response(HTTPStatus.OK)
431 self.send_header("Content-Type", content_type)
432 self.send_header("Content-Length", str(len(body)))
433 self.send_header("Cache-Control", "no-store")
434 self.end_headers()
435 self.wfile.write(body)
436
437 def do_POST(self):
438 if not self.authorized():
439 return
440 try:
441 request = self.read_json()
442 except ValueError:
443 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON body"})
444 return
445 if self.path == "/session":
446 self.handle_session(request)
447 elif self.path == "/delegate":
448 self.handle_delegate(request)
449 elif self.path == "/approve":
450 self.handle_approve(request)
451 else:
452 self.send_error(HTTPStatus.NOT_FOUND)
453
454 def handle_session(self, request):
455 # The vendor's SDP parser needs the offer byte-exact (trailing CRLF too).
456 sdp = request.get("sdp")
457 if not isinstance(sdp, str) or not sdp.strip():
458 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "sdp offer required"})
459 return
460 try:
461 voice = request.get("voice") or self.config.voice
462 if voice not in self.config.voices():
463 self.send_json(HTTPStatus.BAD_REQUEST, {"error": f"unknown voice {voice!r}"})
464 return
465 result = create_live_session(
466 self.config, sdp, live_history(request.get("history") or []), voice
467 )
468 except urllib.error.HTTPError as error:
469 detail = error.read().decode("utf-8", "replace")[:800]
470 log.warning("live session creation failed: %s %s", error.code, detail)
471 self.send_json(
472 HTTPStatus.BAD_GATEWAY,
473 {"error": f"OpenAI rejected the session ({error.code}): {detail}"},
474 )
475 return
476 except (urllib.error.URLError, OSError, KeyError, ValueError) as error:
477 log.warning("live session creation failed: %s", error)
478 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"OpenAI: {error}"})
479 return
480 self.send_json(HTTPStatus.OK, result)
481
482 def handle_delegate(self, request):
483 hermes_session = request.get("hermes_session")
484 lines = request.get("lines")
485 if not isinstance(hermes_session, str) or not SESSION_ID_RE.match(hermes_session):
486 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid hermes_session"})
487 return
488 if not isinstance(lines, list) or not lines:
489 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "lines required"})
490 return
491 # One run per conversation: a newer delegation (usually a correction)
492 # supersedes and stops the previous run.
493 previous = take_run(hermes_session)
494 if previous:
495 stop_run(self.config, previous)
496 try:
497 run_id = hermes_start_run(self.config, hermes_session, lines)
498 except urllib.error.HTTPError as error:
499 detail = error.read().decode("utf-8", "replace")[:400]
500 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes {error.code}: {detail}"})
501 return
502 except (urllib.error.URLError, OSError, ValueError, KeyError) as error:
503 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes unreachable: {error}"})
504 return
505 with RUNS_LOCK:
506 RUNS[hermes_session] = run_id
507 self.send_response(HTTPStatus.OK)
508 self.send_header("Content-Type", "application/x-ndjson")
509 self.send_header("Cache-Control", "no-store")
510 self.send_header("X-Accel-Buffering", "no")
511 self.end_headers()
512 try:
513 try:
514 started = time.monotonic()
515 tools = 0
516 for item in hermes_follow_run(self.config, run_id):
517 if item["type"] == "tool" and item["status"] == "running":
518 tools += 1
519 self.write_line(item)
520 log.info(
521 "hermes turn took %.1fs (%d tool calls)", time.monotonic() - started, tools
522 )
523 except urllib.error.HTTPError as error:
524 detail = error.read().decode("utf-8", "replace")[:400]
525 self.write_line({"type": "error", "message": f"Hermes {error.code}: {detail}"})
526 except (urllib.error.URLError, OSError, ValueError) as error:
527 self.write_line({"type": "error", "message": f"Hermes unreachable: {error}"})
528 take_run(hermes_session, run_id)
529 except (BrokenPipeError, ConnectionResetError):
530 # The page went away; the run keeps going and stays registered, so
531 # a later delegation still supersedes it.
532 log.info("delegation for %s abandoned by the page", hermes_session)
533
534 def handle_approve(self, request):
535 run_id = request.get("run_id")
536 request_id = request.get("request_id")
537 choice = request.get("choice")
538 # No "always": a permanent allowlist entry is not a voice decision.
539 if (
540 choice not in ("once", "session", "deny")
541 or not isinstance(run_id, str)
542 or not RUN_ID_RE.match(run_id)
543 or not (
544 request_id is None
545 or (isinstance(request_id, str) and RUN_ID_RE.match(request_id))
546 )
547 ):
548 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid approval answer"})
549 return
550 try:
551 result = hermes_approve(self.config, run_id, request_id, choice)
552 except urllib.error.HTTPError as error:
553 detail = error.read().decode("utf-8", "replace")[:400]
554 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes {error.code}: {detail}"})
555 return
556 except (urllib.error.URLError, OSError, ValueError) as error:
557 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes unreachable: {error}"})
558 return
559 log.info("approval %s for run %s: %s", choice, run_id, result.get("resolved"))
560 self.send_json(HTTPStatus.OK, result)
561
562 def write_line(self, item):
563 self.wfile.write(json.dumps(item).encode() + b"\n")
564 self.wfile.flush()
565
566
567def main():
568 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
569 parser.add_argument("--listen", default="127.0.0.1:8790", help="host:port to serve on")
570 parser.add_argument("--hermes-url", required=True, help="Hermes API server base URL")
571 parser.add_argument("--hermes-key-file", required=True, help="file holding API_SERVER_KEY")
572 parser.add_argument("--openai-key-file", required=True, help="file holding the OpenAI API key")
573 parser.add_argument("--access-token-file", help="file holding the token the page must present")
574 parser.add_argument("--config", help="JSON file: voice, model, idle_seconds, glossary, instructions")
575 parser.add_argument(
576 "--static",
577 default=os.environ.get("HERMES_VOICE_STATIC"),
578 help="directory with index.html, app.js, style.css (default: $HERMES_VOICE_STATIC)",
579 )
580 args = parser.parse_args()
581 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
582
583 host, _, port = args.listen.rpartition(":")
584 host = host.strip("[]")
585 if args.access_token_file is None and host not in ("127.0.0.1", "::1", "localhost"):
586 parser.error("--access-token-file is required when not listening on loopback")
587
588 raw = json.loads(Path(args.config).read_text()) if args.config else {}
589 config = Config(
590 raw,
591 openai_key=Path(args.openai_key_file).read_text().strip(),
592 hermes_url=args.hermes_url,
593 hermes_key=Path(args.hermes_key_file).read_text().strip(),
594 )
595 Handler.config = config
596 Handler.static = Path(args.static)
597 if args.access_token_file:
598 Handler.access_token = Path(args.access_token_file).read_text().strip()
599 server_class = ThreadingHTTPServer
600 if ":" in host:
601 # "::" is dual-stack on Linux (bindv6only=0).
602 server_class = type("Server6", (ThreadingHTTPServer,), {"address_family": socket.AF_INET6})
603 server = server_class((host, int(port)), Handler)
604 server.daemon_threads = True
605 log.info("serving on http://%s", args.listen)
606 server.serve_forever()
607
608
609if __name__ == "__main__":
610 main()