| """A minimal OpenAI-compatible server standing in for the Cortex.0 runtime. |
| |
| Cortex.0 itself needs ~45 GiB of RAM and a GPU, neither of which this machine |
| has. This server speaks the same protocol so the fusion code can be validated |
| for real: real HTTP, real SDK calls, real agent loop. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import threading |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
|
|
|
|
| class Handler(BaseHTTPRequestHandler): |
| protocol_version = "HTTP/1.1" |
| calls: list[dict] = [] |
|
|
| def log_message(self, *args): |
| pass |
|
|
| def _send(self, code: int, payload: dict) -> None: |
| body = json.dumps(payload).encode() |
| self.send_response(code) |
| self.send_header("Content-Type", "application/json") |
| self.send_header("Content-Length", str(len(body))) |
| self.end_headers() |
| self.wfile.write(body) |
|
|
| def do_GET(self): |
| if self.path.rstrip("/").endswith("/models"): |
| self._send(200, {"object": "list", "data": [{"id": "cortex0", "object": "model"}]}) |
| else: |
| self._send(404, {"error": "not found"}) |
|
|
| def do_POST(self): |
| length = int(self.headers.get("Content-Length", 0)) |
| request = json.loads(self.rfile.read(length) or b"{}") |
| Handler.calls.append(request) |
| messages = request.get("messages", []) |
| tools = request.get("tools") or [] |
|
|
| |
| already_acted = any(m.get("role") == "tool" for m in messages) |
| if tools and not already_acted: |
| self._send(200, { |
| "id": "cmpl-1", "object": "chat.completion", "model": "cortex0", |
| "choices": [{ |
| "index": 0, "finish_reason": "tool_calls", |
| "message": { |
| "role": "assistant", "content": None, |
| "tool_calls": [{ |
| "id": "call_1", "type": "function", |
| "function": {"name": tools[0]["function"]["name"], |
| "arguments": json.dumps({"command": "echo cortex0-fused"})}, |
| }], |
| }, |
| }], |
| "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, |
| }) |
| else: |
| self._send(200, { |
| "id": "cmpl-2", "object": "chat.completion", "model": "cortex0", |
| "choices": [{ |
| "index": 0, "finish_reason": "stop", |
| "message": {"role": "assistant", "content": "Task complete."}, |
| }], |
| "usage": {"prompt_tokens": 20, "completion_tokens": 4, "total_tokens": 24}, |
| }) |
|
|
|
|
| def start(port: int = 0) -> tuple[ThreadingHTTPServer, str]: |
| server = ThreadingHTTPServer(("127.0.0.1", port), Handler) |
| threading.Thread(target=server.serve_forever, daemon=True).start() |
| return server, f"http://127.0.0.1:{server.server_address[1]}/v1" |
|
|
