popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add OpenAI-compatible gateway

Browse Source 
- GET /v1/health, GET /v1/models, POST /v1/chat/completions (streaming + non-streaming)
- OpenAI SDK, LangChain ChatOpenAI, and SSE streaming integration tests
- Tracker-backed GET /v1/models endpoint
- OpenAI-format errors for unavailable model (503) and pipeline failures
- Malformed JSON body handled with 400 instead of crash
- Test deps (openai, langchain-openai) declared in root pyproject dev extras

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dobromir Popov
2026-06-29 01:52:45 +03:00
parent 0199c99f6b
commit 3b7ef2743d
5 changed files with 366 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
packages/gateway/meshnet_gateway/server.py
View File

@@ -10,16 +10,33 @@ import urllib.request
class _GatewayHTTPServer(http.server.HTTPServer):
def __init__(self, addr, handler, inference_route: list[str] | None, tracker_url: str | None):
def __init__(
self,
addr,
handler,
inference_route: list[str] | None,
tracker_url: str | None,
model_presets: list[str] | None,
):
super().__init__(addr, handler)
self.inference_route = inference_route
self.tracker_url = tracker_url
self.model_presets = model_presets
class _GatewayHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass
def do_GET(self):
if self.path == "/v1/health":
self._handle_health()
elif self.path == "/v1/models":
self._handle_models()
else:
self.send_response(404)
self.end_headers()
def do_POST(self):
if self.path == "/v1/chat/completions":
self._handle_chat_completions()
@@ -27,18 +44,57 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
self.send_response(404)
self.end_headers()
def _send_json_error(self, status: int, message: str) -> None:
resp = json.dumps({"error": message}).encode()
def _send_json(self, status: int, data: dict) -> None:
body = json.dumps(data).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(resp)))
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(resp)
self.wfile.write(body)
def _send_json_error(self, status: int, message: str) -> None:
self._send_json(status, {"error": message})
def _send_openai_error(self, status: int, message: str, code: str) -> None:
self._send_json(status, {
"error": {
"message": message,
"type": "invalid_request_error",
"param": "model",
"code": code,
}
})
def _handle_health(self) -> None:
self._send_json(200, {"status": "ok"})
def _handle_models(self) -> None:
server: _GatewayHTTPServer = self.server # type: ignore[assignment]
created = int(time.time())
if server.tracker_url is not None:
try:
models_resp = _get_json(f"{server.tracker_url}/v1/models")
self._send_json(200, models_resp)
return
except Exception:
self._send_json_error(503, "tracker unavailable")
return
presets = server.model_presets or ["stub-model"]
data = [
{"id": p, "object": "model", "created": created, "owned_by": "meshnet"}
for p in presets
]
self._send_json(200, {"object": "list", "data": data})
def _handle_chat_completions(self):
server: _GatewayHTTPServer = self.server # type: ignore[assignment]
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length))
try:
body = json.loads(self.rfile.read(length))
except (json.JSONDecodeError, ValueError):
self._send_openai_error(400, "invalid JSON body", "invalid_request_error")
return
streaming = bool(body.get("stream", False))
if server.tracker_url is not None:
model = body.get("model", "stub-model")
@@ -54,7 +110,11 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
raise ValueError("invalid tracker route")
except urllib.error.HTTPError as exc:
error_body = _safe_error_body(exc)
self._send_json_error(503, error_body.get("error", "tracker unavailable"))
self._send_openai_error(
503,
error_body.get("error", "model not available"),
"model_not_available",
)
return
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, TypeError, ValueError):
self._send_json_error(503, "tracker unavailable")
@@ -64,38 +124,87 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
messages = body.get("messages", [])
# Send messages to the first node in the inference route.
node_resp = _post_json(f"{route[0]}/v1/infer", {"messages": messages})
try:
# Send messages to the first node in the inference route.
node_resp = _post_json(f"{route[0]}/v1/infer", {"messages": messages})
# Forward activations through remaining nodes in the route.
for node_url in route[1:]:
node_resp = _post_json(f"{node_url}/v1/infer", {"activations": node_resp["activations"]})
# Forward activations through remaining nodes in the route.
for node_url in route[1:]:
node_resp = _post_json(f"{node_url}/v1/infer", {"activations": node_resp["activations"]})
except (urllib.error.URLError, TimeoutError, KeyError, json.JSONDecodeError) as exc:
self._send_openai_error(503, f"inference pipeline error: {exc}", "pipeline_error")
return
response_body = json.dumps({
"id": "chatcmpl-stub",
"object": "chat.completion",
"created": int(time.time()),
"model": body.get("model", "stub-model"),
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": node_resp["text"],
text = node_resp["text"]
model = body.get("model", "stub-model")
if streaming:
self._send_sse_response(text, model)
else:
self._send_json(200, {
"id": "chatcmpl-stub",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": text,
},
"finish_reason": "stop",
}],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
},
"finish_reason": "stop",
}],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
},
}).encode()
})
def _send_sse_response(self, text: str, model: str) -> None:
chunk_id = "chatcmpl-stub"
created = int(time.time())
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(response_body)))
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Accel-Buffering", "no")
self.end_headers()
self.wfile.write(response_body)
def _emit(data: str) -> None:
self.wfile.write(f"data: {data}\n\n".encode())
self.wfile.flush()
# Role delta
_emit(json.dumps({
"id": chunk_id,
"object": "chat.completion.chunk",
"created": created,
"model": model,
"choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}],
}))
# Content delta
_emit(json.dumps({
"id": chunk_id,
"object": "chat.completion.chunk",
"created": created,
"model": model,
"choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}],
}))
# Stop delta
_emit(json.dumps({
"id": chunk_id,
"object": "chat.completion.chunk",
"created": created,
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
}))
# SSE sentinel
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
def _safe_error_body(exc: urllib.error.HTTPError) -> dict:
@@ -125,6 +234,9 @@ class GatewayServer:
- ``tracker_url``: gateway queries the tracker per request using the model name.
- ``inference_route``: static ordered list of node base URLs.
- ``node_url``: single-node shortcut (US-001 compat), equivalent to inference_route=[node_url].
Optional ``model_presets`` lists the model names returned by GET /v1/models when using
a static inference route. Ignored when using tracker_url (tracker provides the list).
"""
def __init__(
@@ -133,6 +245,7 @@ class GatewayServer:
*,
inference_route: list[str] | None = None,
tracker_url: str | None = None,
model_presets: list[str] | None = None,
host: str = "127.0.0.1",
port: int = 0,
):
@@ -149,6 +262,7 @@ class GatewayServer:
self._inference_route: list[str] | None = inference_route
self._tracker_url: str | None = tracker_url.rstrip("/") if tracker_url is not None else None
self._model_presets: list[str] | None = model_presets
self._host = host
self._requested_port = port
self._server: _GatewayHTTPServer | None = None
@@ -164,6 +278,7 @@ class GatewayServer:
_GatewayHandler,
self._inference_route,
self._tracker_url,
self._model_presets,
)
self.port = self._server.server_address[1]
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)