Spaces:
Configuration error
Configuration error
| """ | |
| HTTP Facade for ToGMAL MCP (stdio) server | |
| Exposes minimal HTTP endpoints to interact with MCP tools: | |
| - POST /list-tools-dynamic { conversation_history, user_context } | |
| - POST /call-tool { name, arguments } | |
| This wraps the MCP tool functions directly and returns their outputs. | |
| No authentication is implemented; use locally only. | |
| """ | |
| import json | |
| import asyncio | |
| from http.server import BaseHTTPRequestHandler, HTTPServer | |
| from urllib.parse import urlparse | |
| # Import MCP tool functions and input models | |
| from togmal_mcp import ( | |
| togmal_list_tools_dynamic, | |
| get_recommended_checks, | |
| analyze_prompt, | |
| analyze_response, | |
| get_taxonomy, | |
| get_statistics, | |
| togmal_check_prompt_difficulty, | |
| AnalyzePromptInput, | |
| AnalyzeResponseInput, | |
| GetTaxonomyInput, | |
| ResponseFormat, | |
| ) | |
| class MCPHTTPRequestHandler(BaseHTTPRequestHandler): | |
| server_version = "ToGMALHTTP/0.1" | |
| def _write_json(self, status: int, payload: dict): | |
| self.send_response(status) | |
| self.send_header("Content-Type", "application/json") | |
| self.end_headers() | |
| self.wfile.write(json.dumps(payload).encode("utf-8")) | |
| def do_GET(self): | |
| path = urlparse(self.path).path | |
| if path in ("/", "/index.html"): | |
| self.send_response(200) | |
| self.send_header("Content-Type", "text/html") | |
| self.end_headers() | |
| html = """ | |
| <!doctype html> | |
| <html> | |
| <head><title>ToGMAL HTTP Facade</title></head> | |
| <body> | |
| <h1>ToGMAL HTTP Facade</h1> | |
| <p>This server exposes MCP tools over HTTP for local development.</p> | |
| <ul> | |
| <li>POST /list-tools-dynamic - body: {\"conversation_history\": [...], \"user_context\": {...}}</li> | |
| <li>POST /call-tool - body: {\"name\": \"togmal_analyze_prompt\", \"arguments\": {...}}</li> | |
| </ul> | |
| <p>Supported names for /call-tool: togmal_analyze_prompt, togmal_analyze_response, togmal_get_taxonomy, togmal_get_statistics, togmal_list_tools_dynamic, togmal_get_recommended_checks, togmal_check_prompt_difficulty.</p> | |
| </body> | |
| </html> | |
| """ | |
| self.wfile.write(html.encode("utf-8")) | |
| return | |
| if path == "/favicon.ico": | |
| self.send_response(204) | |
| self.end_headers() | |
| return | |
| self.send_error(404, "Not Found") | |
| def do_POST(self): | |
| path = urlparse(self.path).path | |
| try: | |
| content_len = int(self.headers.get("Content-Length", "0")) | |
| except Exception: | |
| content_len = 0 | |
| raw = self.rfile.read(content_len) if content_len > 0 else b"{}" | |
| try: | |
| data = json.loads(raw.decode("utf-8") or "{}") | |
| except Exception: | |
| return self._write_json(400, {"error": "Invalid JSON body"}) | |
| # Ensure an event loop exists for async tool calls | |
| loop = asyncio.new_event_loop() | |
| asyncio.set_event_loop(loop) | |
| try: | |
| if path == "/list-tools-dynamic": | |
| conversation_history = data.get("conversation_history") | |
| user_context = data.get("user_context") | |
| result = loop.run_until_complete( | |
| togmal_list_tools_dynamic(conversation_history, user_context) | |
| ) | |
| # result is a JSON string | |
| try: | |
| return self._write_json(200, json.loads(result)) | |
| except Exception: | |
| return self._write_json(200, {"result": result}) | |
| elif path == "/call-tool": | |
| name = data.get("name") | |
| arguments = data.get("arguments", {}) | |
| if not name: | |
| return self._write_json(400, {"error": "Missing 'name'"}) | |
| # Route to specific tool functions | |
| if name == "togmal_analyze_prompt": | |
| params = AnalyzePromptInput(**arguments) | |
| output = loop.run_until_complete(analyze_prompt(params)) | |
| return self._write_json(200, {"result": output}) | |
| elif name == "togmal_analyze_response": | |
| params = AnalyzeResponseInput(**arguments) | |
| output = loop.run_until_complete(analyze_response(params)) | |
| return self._write_json(200, {"result": output}) | |
| elif name == "togmal_get_taxonomy": | |
| params = GetTaxonomyInput(**arguments) | |
| output = loop.run_until_complete(get_taxonomy(params)) | |
| return self._write_json(200, {"result": output}) | |
| elif name == "togmal_get_statistics": | |
| fmt = arguments.get("response_format", "markdown") | |
| output = loop.run_until_complete(get_statistics(ResponseFormat(fmt))) | |
| return self._write_json(200, {"result": output}) | |
| elif name == "togmal_list_tools_dynamic": | |
| conversation_history = arguments.get("conversation_history") | |
| user_context = arguments.get("user_context") | |
| result = loop.run_until_complete( | |
| togmal_list_tools_dynamic(conversation_history, user_context) | |
| ) | |
| try: | |
| return self._write_json(200, json.loads(result)) | |
| except Exception: | |
| return self._write_json(200, {"result": result}) | |
| elif name == "togmal_get_recommended_checks": | |
| conversation_history = arguments.get("conversation_history") | |
| user_context = arguments.get("user_context") | |
| result = loop.run_until_complete( | |
| get_recommended_checks(conversation_history, user_context) | |
| ) | |
| try: | |
| return self._write_json(200, json.loads(result)) | |
| except Exception: | |
| return self._write_json(200, {"result": result}) | |
| elif name == "togmal_check_prompt_difficulty": | |
| prompt = arguments.get("prompt", "") | |
| k = arguments.get("k", 5) | |
| domain_filter = arguments.get("domain_filter") | |
| result = loop.run_until_complete( | |
| togmal_check_prompt_difficulty(prompt, k, domain_filter) | |
| ) | |
| try: | |
| return self._write_json(200, json.loads(result)) | |
| except Exception: | |
| return self._write_json(200, {"result": result}) | |
| else: | |
| return self._write_json(404, {"error": f"Unknown tool: {name}"}) | |
| else: | |
| return self._write_json(404, {"error": "Not Found"}) | |
| except Exception as e: | |
| return self._write_json(500, {"error": str(e)}) | |
| finally: | |
| try: | |
| loop.close() | |
| except Exception: | |
| pass | |
| def run(port: int = 6274): | |
| server_address = ("127.0.0.1", port) | |
| httpd = HTTPServer(server_address, MCPHTTPRequestHandler) | |
| print(f"HTTP MCP facade listening on http://{server_address[0]}:{server_address[1]}") | |
| httpd.serve_forever() | |
| if __name__ == "__main__": | |
| run() | |