#!/usr/bin/env python3 """ Data Dashboard MCP Server - standard MCP Apps protocol (SEP-1865). - tools/call returns structured data only (no HTML) - resources/read returns static HTML App files - Host renders HTML App in iframe, passes tool data via postMessage """ import asyncio import json import os from typing import Any, Dict from mcp_common import ( create_error_response, create_ping_response, create_tools_list_response, load_tools_from_json, handle_mcp_streaming, ) RESOURCE_MIME_TYPE = "text/html;profile=mcp-app" APPS_DIR = os.path.join(os.path.dirname(__file__), "apps") # Resource URI -> static HTML App file mapping RESOURCE_MAP = { "ui://data-dashboard/metrics": "metrics.html", "ui://data-dashboard/chart": "chart.html", "ui://data-dashboard/multi-chart": "multi-chart.html", } RESOURCE_DEFINITIONS = [ { "uri": "ui://data-dashboard/metrics", "name": "metrics-dashboard", "title": "Metrics Dashboard", "description": "Renders KPI metric cards", "mimeType": RESOURCE_MIME_TYPE, }, { "uri": "ui://data-dashboard/chart", "name": "chart", "title": "Chart", "description": "Renders a single ECharts chart", "mimeType": RESOURCE_MIME_TYPE, }, { "uri": "ui://data-dashboard/multi-chart", "name": "multi-chart", "title": "Multi Chart", "description": "Renders multiple ECharts charts in a grid", "mimeType": RESOURCE_MIME_TYPE, }, ] def _load_app_html(uri: str) -> str: """Load static HTML App file for the given resource URI.""" filename = RESOURCE_MAP.get(uri) if not filename: raise ValueError(f"Unknown resource URI: {uri}") filepath = os.path.join(APPS_DIR, filename) with open(filepath, "r", encoding="utf-8") as f: return f.read() def _create_app_response(resource_uri: str, data: Dict[str, Any], width: str = "100%", height: str = "auto") -> Dict[str, Any]: """Create a tool result for MCP Apps protocol. Returns a JSON payload with type:"app" that tells the host to: 1. Fetch the static HTML App via resources/read(resourceUri) 2. Render it in a sandboxed iframe 3. Send `data` to the iframe via postMessage """ app_json = json.dumps({ "type": "app", "resourceUri": resource_uri, "data": data, "_meta": { "mcpui.dev/ui-preferred-frame-size": [width, height], }, }, ensure_ascii=False) return {"content": [{"type": "text", "text": app_json}]} # --------------------------------------------------------------------------- # Tool handlers — return data only, no HTML generation # --------------------------------------------------------------------------- def _handle_render_metrics(arguments: Dict[str, Any]) -> Dict[str, Any]: title = arguments.get("title", "Dashboard") metrics = arguments.get("metrics", []) if not metrics: raise ValueError("Missing required parameter: metrics") return _create_app_response( "ui://data-dashboard/metrics", {"title": title, "metrics": metrics}, "100%", "auto", ) def _handle_render_chart(arguments: Dict[str, Any]) -> Dict[str, Any]: title = arguments.get("title", "Chart") chart_type = arguments.get("chart_type") chart_data = arguments.get("data", {}) if not chart_type: raise ValueError("Missing required parameter: chart_type") if not chart_data or not chart_data.get("series"): raise ValueError("Missing required parameter: data.series") width = arguments.get("width", "100%") height = arguments.get("height", "400px") return _create_app_response( "ui://data-dashboard/chart", { "title": title, "chart_type": chart_type, "data": chart_data, "theme": arguments.get("theme", "light"), "stacked": arguments.get("stacked", False), "smooth": arguments.get("smooth", False), "show_label": arguments.get("show_label", False), }, width, height, ) def _handle_render_multi_chart(arguments: Dict[str, Any]) -> Dict[str, Any]: title = arguments.get("title", "Dashboard") charts = arguments.get("charts", []) if not charts: raise ValueError("Missing required parameter: charts") for i, c in enumerate(charts): if not c.get("chart_type"): raise ValueError(f"charts[{i}]: missing chart_type") if not c.get("data", {}).get("series"): raise ValueError(f"charts[{i}]: missing data.series") columns = arguments.get("columns", 2) num_rows = (len(charts) + columns - 1) // columns total_height = f"{80 + num_rows * 420}px" return _create_app_response( "ui://data-dashboard/multi-chart", { "title": title, "charts": charts, "columns": columns, "theme": arguments.get("theme", "light"), }, "100%", total_height, ) TOOL_HANDLERS = { "render_metrics": _handle_render_metrics, "render_chart": _handle_render_chart, "render_multi_chart": _handle_render_multi_chart, } async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: """Handle an MCP request.""" try: method = request.get("method") params = request.get("params", {}) request_id = request.get("id") if method == "initialize": return { "jsonrpc": "2.0", "id": request_id, "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {}, "resources": {}, }, "serverInfo": { "name": "data-dashboard", "version": "3.0.0", }, }, } elif method == "ping": return create_ping_response(request_id) elif method == "tools/list": tools = load_tools_from_json("dashboard_tools.json") if not tools: tools = [ { "name": name, "description": f"Render {name.replace('render_', '')}", "inputSchema": {"type": "object", "properties": {}, "required": []}, "_meta": {"ui": {"resourceUri": uri}}, } for name, uri in [ ("render_metrics", "ui://data-dashboard/metrics"), ("render_chart", "ui://data-dashboard/chart"), ("render_multi_chart", "ui://data-dashboard/multi-chart"), ] ] return create_tools_list_response(request_id, tools) elif method == "resources/list": return { "jsonrpc": "2.0", "id": request_id, "result": {"resources": RESOURCE_DEFINITIONS}, } elif method == "resources/read": uri = params.get("uri", "") try: html = _load_app_html(uri) except (ValueError, FileNotFoundError) as e: return create_error_response(request_id, -32602, str(e)) return { "jsonrpc": "2.0", "id": request_id, "result": { "contents": [ { "uri": uri, "mimeType": RESOURCE_MIME_TYPE, "text": html, } ] }, } elif method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) handler = TOOL_HANDLERS.get(tool_name) if not handler: return create_error_response(request_id, -32601, f"Unknown tool: {tool_name}") try: result = handler(arguments) except ValueError as e: return create_error_response(request_id, -32602, str(e)) return {"jsonrpc": "2.0", "id": request_id, "result": result} else: return create_error_response(request_id, -32601, f"Unknown method: {method}") except Exception as e: return create_error_response( request.get("id"), -32603, f"Internal error: {str(e)}" ) async def main(): await handle_mcp_streaming(handle_request) if __name__ == "__main__": asyncio.run(main())