"""Small stdio MCP facade for the authenticated Nelloz model-tool mesh.""" from __future__ import annotations import json import os import sys import urllib.error import urllib.request from typing import Any, Callable, Mapping from uuid import uuid4 _TOOLS = { "nelloz_reason": "reasoning.delegate", "nelloz_search": "internet.search", "nelloz_image": "image.generate", "nelloz_edit_image": "image.edit", "nelloz_video": "video.generate", "nelloz_audio": "audio.generate", "nelloz_speak": "voice.speak", } def request_model_tool(payload: Mapping[str, Any]) -> tuple[int, dict[str, Any]]: base = os.getenv("OMNI_CONDUCTOR_URL", "http://127.0.0.1:8100").rstrip("/") token = os.getenv("OMNI_CONDUCTOR_TOKEN", "") if not token: raise RuntimeError("OMNI_CONDUCTOR_TOKEN is required") body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") request = urllib.request.Request( f"{base}/v1/model-tools", data=body, method="POST", headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Idempotency-Key": f"mcp:{payload['session_id']}:{payload['turn_id']}:{payload['tool']}", }, ) try: with urllib.request.urlopen(request, timeout=30) as response: return response.status, json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: raw = exc.read(1_048_576) try: value = json.loads(raw.decode("utf-8")) except (UnicodeError, json.JSONDecodeError): value = {"detail": f"HTTP {exc.code}"} return exc.code, value class ModelToolsMcpServer: def __init__( self, *, requester: Callable[[Mapping[str, Any]], tuple[int, dict[str, Any]]] = request_model_tool, ) -> None: self._requester = requester def handle(self, message: Mapping[str, Any]) -> dict[str, Any] | None: request_id = message.get("id") if request_id is None: return None try: method = str(message.get("method") or "") if method == "initialize": params = message.get("params") if isinstance(message.get("params"), Mapping) else {} result: dict[str, Any] = { "protocolVersion": str(params.get("protocolVersion") or "2025-11-25"), "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "nelloz-model-mesh", "version": "1.0"}, } elif method == "ping": result = {} elif method == "tools/list": result = {"tools": _tool_descriptors()} elif method == "tools/call": result = self._call(message.get("params")) else: return _error(request_id, -32601, "Method not found") return {"jsonrpc": "2.0", "id": request_id, "result": result} except ValueError as exc: return _error(request_id, -32602, str(exc)) except Exception as exc: return _error(request_id, -32603, f"Tool failed: {type(exc).__name__}") def _call(self, value: object) -> dict[str, Any]: if not isinstance(value, Mapping): raise ValueError("tools/call params must be an object") name = str(value.get("name") or "") tool = _TOOLS.get(name) if tool is None: raise ValueError("unknown Nelloz model tool") arguments = value.get("arguments", {}) if not isinstance(arguments, Mapping): raise ValueError("tool arguments must be an object") prompt = str(arguments.get("prompt") or "").strip() if not prompt: raise ValueError("prompt is required") options = arguments.get("options", {}) if not isinstance(options, Mapping): raise ValueError("options must be an object") if tool == "image.edit": artifacts = arguments.get("input_artifacts") if not isinstance(artifacts, list) or not artifacts: raise ValueError("input_artifacts is required for image editing") options = {**options, "input_artifacts": artifacts} payload = { "source_model": str(arguments.get("source_model") or "qwen"), "tool": tool, "session_id": str(arguments.get("session_id") or f"mcp_{uuid4().hex}"), "turn_id": str(arguments.get("turn_id") or f"turn_{uuid4().hex}"), "prompt": prompt, "options": dict(options), } status, response = self._requester(payload) return { "content": [{"type": "text", "text": json.dumps(response, ensure_ascii=False)}], "structuredContent": response, "isError": not 200 <= status < 300, } def _tool_descriptors() -> list[dict[str, Any]]: descriptions = { "nelloz_reason": "Delegate a difficult reasoning or agent task to always-warm Qwen.", "nelloz_search": "Ask Qwen/Hermes to perform an authenticated internet search.", "nelloz_image": "Generate an image with cold-loaded Hunyuan NF4 v2.", "nelloz_edit_image": "Edit one or more artifact-referenced images with Hunyuan.", "nelloz_video": "Generate audiovisual video with cold-loaded MiniMax H3 FL2VA.", "nelloz_audio": "Generate an audio-only artifact through MiniMax H3's audio path.", "nelloz_speak": "Speak text through the always-warm local MiniCPM voice session.", } result = [] for name in _TOOLS: properties: dict[str, Any] = { "prompt": {"type": "string", "minLength": 1}, "session_id": {"type": "string"}, "turn_id": {"type": "string"}, "source_model": { "type": "string", "enum": ["minicpm", "qwen", "hermes", "conductor", "connector"], "default": "qwen", }, "options": {"type": "object", "additionalProperties": True}, } required = ["prompt"] if name == "nelloz_edit_image": properties["input_artifacts"] = { "type": "array", "minItems": 1, "items": {"type": "string"} } required.append("input_artifacts") result.append({ "name": name, "description": descriptions[name], "inputSchema": { "type": "object", "properties": properties, "required": required, "additionalProperties": False, }, }) return result def _error(request_id: object, code: int, message: str) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message[:240]}} def main() -> int: server = ModelToolsMcpServer() for line in sys.stdin: try: message = json.loads(line) if not isinstance(message, dict): continue response = server.handle(message) if response is not None: sys.stdout.write(json.dumps(response, ensure_ascii=False, separators=(",", ":")) + "\n") sys.stdout.flush() except (UnicodeError, json.JSONDecodeError): continue return 0 if __name__ == "__main__": raise SystemExit(main())