#!/usr/bin/env python3 """ Data Dashboard MCP Server - renders metric data as an interactive dashboard. Returns UIResource via the mcp-ui protocol so the frontend renders it as an iframe. """ import asyncio import json from typing import Any, Dict, List from mcp_ui_server import create_ui_resource, UIMetadataKey from mcp_common import ( create_error_response, create_initialize_response, create_ping_response, create_tools_list_response, load_tools_from_json, handle_mcp_streaming, ) def _build_dashboard_html(title: str, metrics: List[Dict[str, str]]) -> str: """Build a self-contained HTML dashboard from metrics data.""" cards_html = "" for m in metrics: label = _esc(m.get("label", "")) value = _esc(m.get("value", "")) change = m.get("change", "") change_type = m.get("change_type", "neutral") change_html = "" if change: color = "#16a34a" if change_type == "up" else "#dc2626" if change_type == "down" else "#6b7280" arrow = "▲" if change_type == "up" else "▼" if change_type == "down" else "—" change_html = f'{arrow} {_esc(change)}' cards_html += f"""
{label}
{value}
{f'
{change_html}
' if change_html else ''}
""" return f""" {_esc(title)}

{_esc(title)}

{cards_html}
""" def _esc(text: str) -> str: """Minimal HTML escaping.""" return text.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) def render_dashboard(title: str, metrics: List[Dict[str, str]]) -> Dict[str, Any]: """Create a UIResource dashboard and serialize it as JSON for TOOL_RESPONSE.""" try: html = _build_dashboard_html(title, metrics) uri_slug = title.replace(" ", "-").lower()[:50] ui_resource = create_ui_resource( { "uri": f"ui://data-dashboard/{uri_slug}", "content": {"type": "rawHtml", "htmlString": html}, "encoding": "text", "uiMetadata": { UIMetadataKey.PREFERRED_FRAME_SIZE: ["100%", "auto"], }, } ) resource_json = json.dumps( ui_resource.model_dump(mode="json"), ensure_ascii=False ) return {"content": [{"type": "text", "text": resource_json}]} except Exception as e: return { "content": [{"type": "text", "text": f"Error creating dashboard: {str(e)}"}] } 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 create_initialize_response(request_id, "data-dashboard") 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": "render_dashboard", "description": "Render a data dashboard with metric cards.", "inputSchema": { "type": "object", "properties": { "title": {"type": "string"}, "metrics": { "type": "array", "items": { "type": "object", "properties": { "label": {"type": "string"}, "value": {"type": "string"}, "change": {"type": "string"}, "change_type": {"type": "string", "enum": ["up", "down", "neutral"]}, }, "required": ["label", "value"], }, }, }, "required": ["title", "metrics"], }, } ] return create_tools_list_response(request_id, tools) elif method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) if tool_name == "render_dashboard": title = arguments.get("title", "Dashboard") metrics = arguments.get("metrics", []) if not metrics: return create_error_response( request_id, -32602, "Missing required parameter: metrics" ) result = render_dashboard(title, metrics) return {"jsonrpc": "2.0", "id": request_id, "result": result} else: return create_error_response( request_id, -32601, f"Unknown tool: {tool_name}" ) 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(): """Main entry point.""" await handle_mcp_streaming(handle_request) if __name__ == "__main__": asyncio.run(main())