jonny (nonvenomous)

here's the MWE for plain inference as plaintext and highlighted picture of text: import json, socket, struct, uuid SOCK = "/run/hatch/sandbox/space-inference.sock" def call(req, timeout=60): payload = json.dumps(req).encode() s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.settimeout(timeout) s.connect(SOCK) s.sendall(struct.pack(">I", len(payload)) + payload) (n,) = struct.unpack(">I", s.recv(4)) body = b"" while len(body) < n: chunk = s.recv(n - len(body)) if not chunk: raise RuntimeError("truncated response") body += chunk s.close() return json.loads(body) resp = call({ "kind": "complete", "request_id": f"mwe-{uuid.uuid4().hex[:8]}", "slug": "__mwe__", "system": "Always answer in exactly three words.", "prompt": "What is the capital of France?", "json_schema": { "type": "object", "properties": {"reply": {"type": "string"}}, "required": ["reply"], "additionalProperties": False, }, }) print(resp["ok"]) print(resp["result"]["content"]) print(resp["result"].get("usage"))

Here is the MWE for spawning agents, in plaintext and pictures of text: import json, socket, struct, time, uuid SOCK = "/run/hatch/sandbox/space-inference.sock" TOOL_REPORT_PROMPT = ( "List EVERY tool function available to you, exhaustively, one per line " "as namespace.function (for example: muse.exec)." "Do NOT call any tools -- only report the two lists. " "End your reply with exactly: DONE" ) def _roundtrip(req, timeout=120): payload = json.dumps(req).encode() s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.settimeout(timeout) s.connect(SOCK) s.sendall(struct.pack(">I", len(payload)) + payload) (n,) = struct.unpack(">I", _recvall(s, 4)) body = _recvall(s, n) s.close() return json.loads(body) def _recvall(s, n): buf = b"" while len(buf) < n: chunk = s.recv(n - len(buf)) if not chunk: raise RuntimeError("socket closed / truncated response") buf += chunk return buf def spawn(message, slug="__mwe__", poll_interval=10): uid = uuid.uuid4().hex[:8] spawn_resp = _roundtrip({ "kind": "agent_send", "request_id": f"mwe-spawn-{uid}", "slug": slug, "action": "do-whatever-i-say", "invocation_id": f"mwe-inv-{uid}", "spawn_request_id": f"mwe:spawn:{uid}", "message": message, "dedupe_key": f"mwe-{uid}", # with allow_parallel, deduplication "allow_parallel": True, "root_request_id": f"mwe-root-{uid}", }) result = spawn_resp.get("result", {}) if isinstance(spawn_resp, dict) else {} task_id = result.get("task_id") or result.get("space_task_id") if not task_id: raise RuntimeError(f"spawn returned no task_id: {json.dumps(spawn_resp)[:500]}") messages = [] while True: time.sleep(poll_interval) st = _roundtrip({ "kind": "agent_status", "request_id": f"mwe-status-{uid}-{int(time.time())}", "slug": slug, "task_id": task_id, }).get("result", {}) messages.append(st) status = st.get("status") or st.get("state") if status in ("completed", "failed", "terminal"): return { "task_id": task_id, "messages": messages, "status": status, "final_response": st.get("final_response"), } if __name__ == "__main__": run = spawn(TOOL_REPORT_PROMPT) print(f"task {run['task_id']} -> {run['status']} ({len(run['messages'])} snapshots)") print("=" * 60) print(run["final_response"] or "(empty final response)")