1#!/usr/bin/env python32"""Small deterministic OpenAI-compatible model used by the integration lab.34It deliberately implements only the HTTP surface Hermes needs. Scenario5markers are read from the latest user turn, so setup prompts and older turns6cannot accidentally trigger a lab interaction.7"""89from __future__ import annotations1011import json12import re13import time14import uuid15from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer16from typing import Any1718PORT = 808019MARKER = re.compile(r"\b(LAB_(?:HELLO|QUESTION|CUSTOM|OPEN|MULTI|APPROVAL|MENTION|MEDIA)_[A-Za-z0-9_-]+)\b")20MENTION = re.compile(r"@([0-9a-fA-F]{64})")21ATTACHMENT = re.compile(r"\b(tinyagent-attachment-[A-Za-z0-9._-]+)\b")222324def _text(content: Any) -> str:25 if isinstance(content, str):26 return content27 if isinstance(content, list):28 return " ".join(_text(part.get("text", "")) for part in content if isinstance(part, dict))29 return ""303132def _scenario(messages: list[dict[str, Any]]) -> tuple[str | None, int | None]:33 """Return the marker and the index of its user turn, if any."""34 for index in range(len(messages) - 1, -1, -1):35 message = messages[index]36 if message.get("role") != "user":37 continue38 found = MARKER.search(_text(message.get("content")))39 if found:40 return found.group(1), index41 return None, None42 return None, None434445def _latest_user_text(messages: list[dict[str, Any]], start: int) -> str:46 return _text(messages[start].get("content"))474849def _tool_names(tools: list[dict[str, Any]]) -> set[str]:50 names: set[str] = set()51 for tool in tools:52 function = tool.get("function") if isinstance(tool, dict) else None53 if isinstance(function, dict) and isinstance(function.get("name"), str):54 names.add(function["name"])55 return names565758def _tool_result(messages: list[dict[str, Any]], start: int) -> str | None:59 for message in messages[start + 1 :]:60 if message.get("role") == "tool":61 return _text(message.get("content"))62 return None636465def _response(message: dict[str, Any], *, model: str, stream: bool) -> dict[str, Any]:66 response_id = "chatcmpl-" + uuid.uuid4().hex67 choice = {"index": 0, "message": message, "finish_reason": "tool_calls" if message.get("tool_calls") else "stop"}68 return {"id": response_id, "object": "chat.completion", "created": int(time.time()), "model": model,69 "choices": [choice], "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}}707172def complete(body: dict[str, Any]) -> dict[str, Any]:73 messages = body.get("messages") or []74 tools = _tool_names(body.get("tools") or [])75 marker, start = _scenario(messages)76 model = str(body.get("model") or "tiny-lab")77 if not marker or start is None:78 return _response({"role": "assistant", "content": "LAB_IDLE"}, model=model, stream=False)79 result = _tool_result(messages, start)80 if result is not None:81 media = re.search(r"It is saved at: (\S+)\. Its content", _latest_user_text(messages, start)) if marker.startswith("LAB_MEDIA_") else None82 suffix = f"\nMEDIA:{media.group(1)}" if media else ""83 return _response({"role": "assistant", "content": f"LAB_DONE {marker} {result}{suffix}"}, model=model, stream=False)84 if marker.startswith("LAB_HELLO_"):85 return _response({"role": "assistant", "content": f"LAB_DONE {marker}"}, model=model, stream=False)86 latest_text = _latest_user_text(messages, start)87 if marker.startswith("LAB_MENTION_"):88 mentions = MENTION.findall(latest_text)89 suffix = " @" + mentions[0] if mentions else ""90 return _response({"role": "assistant", "content": f"LAB_DONE {marker}{suffix}"}, model=model, stream=False)91 if marker.startswith("LAB_MEDIA_"):92 attachment = ATTACHMENT.search(latest_text)93 cached = re.search(r"It is saved at: (\S+)\. Its content", latest_text)94 if cached and "read_file" in tools:95 call = {"id": "call_" + marker.lower(), "type": "function", "function": {96 "name": "read_file", "arguments": json.dumps({"path": cached.group(1)})}}97 return _response({"role": "assistant", "content": None, "tool_calls": [call]}, model=model, stream=False)98 suffix = f" MEDIA_RECEIVED {attachment.group(1)}" if attachment else " MEDIA_NOT_FOUND"99 return _response({"role": "assistant", "content": f"LAB_DONE {marker}{suffix}"}, model=model, stream=False)100 if marker.startswith(("LAB_QUESTION_", "LAB_MULTI_", "LAB_CUSTOM_", "LAB_OPEN_")):101 if "clarify" not in tools:102 return _response({"role": "assistant", "content": "LAB_ERROR clarify tool is unavailable"}, model=model, stream=False)103 multi = marker.startswith("LAB_MULTI_")104 choices = [] if marker.startswith("LAB_OPEN_") else ["Continue", "Stop"]105 arguments = {"questions": [{"question": f"Choose for {marker}", "choices": choices, "multi_select": multi}]}106 call = {"id": "call_" + marker.lower(), "type": "function", "function": {"name": "clarify", "arguments": json.dumps(arguments)}}107 return _response({"role": "assistant", "content": None, "tool_calls": [call]}, model=model, stream=False)108 if "execute_code" not in tools:109 return _response({"role": "assistant", "content": "LAB_ERROR execute_code tool is unavailable"}, model=model, stream=False)110 call = {"id": "call_" + marker.lower(), "type": "function", "function": {"name": "execute_code", "arguments": json.dumps({"code": "print('tinyagent-lab')"})}}111 return _response({"role": "assistant", "content": None, "tool_calls": [call]}, model=model, stream=False)112113114def _sse_payload(result: dict[str, Any]) -> list[str]:115 choice = result["choices"][0]116 message = choice["message"]117 chunks: list[dict[str, Any]] = []118 base = {"id": result["id"], "object": "chat.completion.chunk", "created": result["created"], "model": result["model"]}119 delta: dict[str, Any] = {"role": "assistant"}120 if message.get("content") is not None:121 delta["content"] = message["content"]122 if message.get("tool_calls"):123 delta["tool_calls"] = [{"index": 0, **message["tool_calls"][0]}]124 chunks.append({**base, "choices": [{"index": 0, "delta": delta, "finish_reason": None}]})125 chunks.append({**base, "choices": [{"index": 0, "delta": {}, "finish_reason": choice["finish_reason"]}]})126 return [f"data: {json.dumps(chunk, separators=(',', ':'))}\n\n" for chunk in chunks] + ["data: [DONE]\n\n"]127128129class Handler(BaseHTTPRequestHandler):130 server_version = "tinyagent-lab/1"131132 def log_message(self, _format: str, *_args: Any) -> None:133 return134135 def do_GET(self) -> None: # noqa: N802136 if self.path == "/healthz":137 self._send(200, {"status": "ok"})138 elif self.path == "/v1/models":139 self._send(200, {"object": "list", "data": [{"id": "tiny-lab", "object": "model", "owned_by": "tinyagent-lab"}]})140 else:141 self._send(404, {"error": {"message": "not found", "type": "invalid_request_error"}})142143 def do_POST(self) -> None: # noqa: N802144 if self.path != "/v1/chat/completions":145 self._send(404, {"error": {"message": "not found", "type": "invalid_request_error"}})146 return147 length = int(self.headers.get("Content-Length", "0"))148 try:149 body = json.loads(self.rfile.read(length))150 result = complete(body)151 except (ValueError, TypeError, json.JSONDecodeError) as exc:152 self._send(400, {"error": {"message": f"invalid JSON: {exc}", "type": "invalid_request_error"}})153 return154 if body.get("stream"):155 payload = "".join(_sse_payload(result)).encode()156 self.send_response(200)157 self.send_header("Content-Type", "text/event-stream")158 self.send_header("Cache-Control", "no-cache")159 self.send_header("Content-Length", str(len(payload)))160 self.end_headers()161 self.wfile.write(payload)162 else:163 self._send(200, result)164165 def _send(self, status: int, value: dict[str, Any]) -> None:166 payload = json.dumps(value, separators=(",", ":")).encode()167 self.send_response(status)168 self.send_header("Content-Type", "application/json")169 self.send_header("Content-Length", str(len(payload)))170 self.end_headers()171 self.wfile.write(payload)172173174def serve(port: int = PORT) -> None:175 ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()176177178if __name__ == "__main__":179 serve()