File size: 4,719 Bytes
12496fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """Single-inference local HTTP endpoint with bounded input and SSE text streaming.
Local development server, not an internet-facing production service. No tool execution.
"""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import hmac
import json
import threading
def create_server(backend, token, port=8765):
if len(token) < 16:
raise ValueError("Use a local API token of at least 16 characters")
active = threading.Lock()
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, *args):
pass
def response(self, status, value):
body = json.dumps(value).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def do_GET(self):
if self.path == "/health":
self.response(200, {"status": "ready", "scope": "local inference; no execution"})
else:
self.response(404, {"error": "not found"})
def do_POST(self):
self.connection.settimeout(30)
if self.path != "/v1/chat/completions":
return self.response(404, {"error": "not found"})
if self.headers.get("Origin"):
return self.response(403, {"error": "Browser cross-origin requests disabled"})
if not hmac.compare_digest(self.headers.get("Authorization", ""), "Bearer " + token):
return self.response(401, {"error": "authentication required"})
if self.headers.get("Transfer-Encoding"):
return self.response(400, {"error": "chunked request bodies unsupported"})
try:
length = int(self.headers.get("Content-Length", "0"))
if not 0 < length <= 65536:
return self.response(413, {"error": "request size limit"})
request = json.loads(self.rfile.read(length))
messages = request["messages"]
if not isinstance(messages, list) or not 1 <= len(messages) <= 64:
raise ValueError("message count")
if any(not isinstance(m, dict) or set(m) != {"role", "content"} or m["role"] not in {"system", "user", "assistant"} or not isinstance(m["content"], str) for m in messages):
raise ValueError("message schema")
if request.get("response_format") or request.get("tools"):
return self.response(400, {"error": "This local endpoint does not implement constrained JSON or native tool calls"})
streaming = request.get("stream", False)
if not isinstance(streaming, bool):
raise ValueError("stream must be boolean")
except (ValueError, TypeError, KeyError):
return self.response(400, {"error": "invalid request"})
if not active.acquire(blocking=False):
return self.response(503, {"error": "inference worker busy"})
started = False
try:
if streaming:
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-store")
self.send_header("Connection", "close")
self.end_headers()
self.close_connection = True
started = True
for chunk in backend.stream(messages):
event = {"choices": [{"index": 0, "delta": {"content": chunk}}]}
self.wfile.write(("data: " + json.dumps(event) + "\n\n").encode())
self.wfile.flush()
self.wfile.write(b"data: [DONE]\n\n")
else:
text = backend.complete(messages)
self.response(200, {"choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}]})
except (BrokenPipeError, ConnectionResetError):
pass
except Exception:
if not started:
self.response(500, {"error": "generation failed; inspect backend locally"})
else:
self.wfile.write(b'data: {"error":"generation failed"}\n\n')
finally:
active.release()
server = ThreadingHTTPServer(("127.0.0.1", port), Handler)
server.daemon_threads = True
return server
|