#!/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('"', """)
# ---------------------------------------------------------------------------
# Chart HTML builder (ECharts)
# ---------------------------------------------------------------------------
ECHARTS_CDN = "https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js"
def _build_chart_html(
title: str,
chart_type: str,
data: Dict[str, Any],
width: str = "100%",
height: str = "400px",
theme: str = "light",
stacked: bool = False,
smooth: bool = False,
show_label: bool = False,
) -> str:
"""Build a self-contained HTML page with an ECharts chart."""
series = data.get("series", [])
categories = data.get("categories", [])
option = _build_echarts_option(
title, chart_type, categories, series,
stacked=stacked, smooth=smooth, show_label=show_label,
)
option_json = json.dumps(option, ensure_ascii=False)
bg_color = "#1a1a2e" if theme == "dark" else "#f8fafc"
text_color = "#e0e0e0" if theme == "dark" else "#0f172a"
echarts_theme = "'dark'" if theme == "dark" else "null"
return f"""
{_esc(title)}
"""
def _build_echarts_option(
title: str,
chart_type: str,
categories: List,
series_data: List[Dict[str, Any]],
stacked: bool = False,
smooth: bool = False,
show_label: bool = False,
) -> Dict[str, Any]:
"""Build an ECharts option dict based on chart_type."""
builders = {
"line": _option_line,
"bar": _option_bar,
"pie": _option_pie,
"radar": _option_radar,
"scatter": _option_scatter,
"gauge": _option_gauge,
}
builder = builders.get(chart_type, _option_line)
return builder(title, categories, series_data, stacked=stacked, smooth=smooth, show_label=show_label)
def _base_option(title: str) -> Dict[str, Any]:
"""Shared base option for all chart types."""
return {
"title": {"text": title, "left": "center", "top": 8, "textStyle": {"fontSize": 16, "fontWeight": 600}},
"tooltip": {"trigger": "axis"},
"legend": {"top": 36},
"grid": {"top": 80, "left": 60, "right": 30, "bottom": 40},
"color": ["#5470c6", "#91cc75", "#fac858", "#ee6666", "#73c0de", "#3ba272", "#fc8452", "#9a60b4"],
}
def _label_cfg(show: bool) -> Dict[str, Any]:
return {"show": show, "position": "top", "fontSize": 11}
def _option_line(title, categories, series_data, stacked=False, smooth=False, show_label=False):
opt = _base_option(title)
opt["tooltip"]["trigger"] = "axis"
opt["xAxis"] = {"type": "category", "data": categories, "boundaryGap": False}
opt["yAxis"] = {"type": "value"}
opt["series"] = [
{
"name": s["name"],
"type": "line",
"data": s["data"],
"smooth": smooth,
"stack": "total" if stacked else None,
"areaStyle": {} if stacked else None,
"label": _label_cfg(show_label),
}
for s in series_data
]
return opt
def _option_bar(title, categories, series_data, stacked=False, smooth=False, show_label=False):
opt = _base_option(title)
opt["xAxis"] = {"type": "category", "data": categories}
opt["yAxis"] = {"type": "value"}
opt["series"] = [
{
"name": s["name"],
"type": "bar",
"data": s["data"],
"stack": "total" if stacked else None,
"label": _label_cfg(show_label),
"barMaxWidth": 50,
}
for s in series_data
]
return opt
def _option_pie(title, categories, series_data, stacked=False, smooth=False, show_label=False):
opt = _base_option(title)
del opt["grid"]
opt["tooltip"] = {"trigger": "item", "formatter": "{b}: {c} ({d}%)"}
# Pie uses the first series; data items are {name, value}
pie_data = series_data[0]["data"] if series_data else []
opt["series"] = [
{
"name": series_data[0]["name"] if series_data else title,
"type": "pie",
"radius": ["40%", "70%"],
"center": ["50%", "55%"],
"data": pie_data,
"emphasis": {"itemStyle": {"shadowBlur": 10, "shadowOffsetX": 0, "shadowColor": "rgba(0,0,0,0.5)"}},
"label": {"show": True, "formatter": "{b}: {d}%"},
"itemStyle": {"borderRadius": 6, "borderColor": "#fff", "borderWidth": 2},
}
]
return opt
def _option_radar(title, categories, series_data, stacked=False, smooth=False, show_label=False):
opt = _base_option(title)
del opt["grid"]
opt["tooltip"] = {"trigger": "item"}
# Determine max value per indicator from all series
max_vals = [0] * len(categories)
for s in series_data:
for i, v in enumerate(s["data"]):
if i < len(max_vals) and v > max_vals[i]:
max_vals[i] = v
indicator = [{"name": c, "max": int(m * 1.2) or 100} for c, m in zip(categories, max_vals)]
opt["radar"] = {"indicator": indicator, "center": ["50%", "55%"]}
opt["series"] = [
{
"type": "radar",
"data": [{"value": s["data"], "name": s["name"]} for s in series_data],
"areaStyle": {"opacity": 0.15},
}
]
return opt
def _option_scatter(title, categories, series_data, stacked=False, smooth=False, show_label=False):
opt = _base_option(title)
opt["tooltip"] = {"trigger": "item", "formatter": "{a}: ({c})"}
opt["xAxis"] = {"type": "value", "scale": True}
opt["yAxis"] = {"type": "value", "scale": True}
opt["series"] = [
{
"name": s["name"],
"type": "scatter",
"data": s["data"],
"symbolSize": 10,
"label": _label_cfg(show_label),
}
for s in series_data
]
return opt
def _option_gauge(title, categories, series_data, stacked=False, smooth=False, show_label=False):
opt = _base_option(title)
del opt["grid"]
opt["tooltip"] = {"trigger": "item"}
# Gauge uses first series, first data value
value = 0
name = title
if series_data:
name = series_data[0]["name"]
data_arr = series_data[0]["data"]
if data_arr:
value = data_arr[0] if isinstance(data_arr[0], (int, float)) else 0
opt["series"] = [
{
"type": "gauge",
"center": ["50%", "60%"],
"startAngle": 200,
"endAngle": -20,
"min": 0,
"max": 100,
"detail": {"formatter": "{value}%", "fontSize": 24, "offsetCenter": [0, "60%"]},
"data": [{"value": value, "name": name}],
"axisLine": {"lineStyle": {"width": 20}},
"progress": {"show": True, "width": 20},
"pointer": {"show": True},
}
]
return opt
# ---------------------------------------------------------------------------
# Multi-chart HTML builder
# ---------------------------------------------------------------------------
def _build_multi_chart_html(
title: str,
charts: List[Dict[str, Any]],
columns: int = 2,
theme: str = "light",
) -> str:
"""Build a single HTML page containing multiple ECharts in a grid layout."""
bg_color = "#1a1a2e" if theme == "dark" else "#f8fafc"
text_color = "#e0e0e0" if theme == "dark" else "#0f172a"
card_bg = "#252547" if theme == "dark" else "#ffffff"
border_color = "#3a3a5c" if theme == "dark" else "#e2e8f0"
echarts_theme = "'dark'" if theme == "dark" else "null"
# Build chart divs and init scripts
chart_divs = ""
chart_scripts = ""
for idx, chart in enumerate(charts):
chart_id = f"chart_{idx}"
chart_title = chart.get("title", f"Chart {idx + 1}")
chart_type = chart.get("chart_type", "line")
chart_data = chart.get("data", {})
chart_height = chart.get("height", "350px")
stacked = chart.get("stacked", False)
smooth = chart.get("smooth", False)
show_label = chart.get("show_label", False)
categories = chart_data.get("categories", [])
series = chart_data.get("series", [])
option = _build_echarts_option(
chart_title, chart_type, categories, series,
stacked=stacked, smooth=smooth, show_label=show_label,
)
option_json = json.dumps(option, ensure_ascii=False)
chart_divs += f"""
"""
chart_scripts += f"""
(function() {{
var el = document.getElementById('{chart_id}');
var c = echarts.init(el, {echarts_theme});
c.setOption({option_json});
charts.push(c);
}})();"""
# Calculate total height based on rows
num_rows = (len(charts) + columns - 1) // columns
# 350px default chart + 32px card padding + 16px gap
est_row_height = 400
total_height = 80 + num_rows * est_row_height
return f"""
{_esc(title)}
{_esc(title)}
{chart_divs}
"""
def _create_ui_resource(uri: str, html: str, width: str = "100%", height: str = "auto") -> Dict[str, Any]:
"""Create a self-contained UIResource with HTML in resource.text."""
ui_resource = create_ui_resource({
"uri": uri,
"content": {"type": "rawHtml", "htmlString": html},
"encoding": "text",
"uiMetadata": {
UIMetadataKey.PREFERRED_FRAME_SIZE: [width, height],
},
})
resource_json = json.dumps(ui_resource.model_dump(mode="json"), ensure_ascii=False)
return {"content": [{"type": "text", "text": resource_json}]}
def _handle_data_viz(uri: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""Unified handler for all data visualization URIs.
Parses URI path to route to the appropriate HTML builder.
All content is self-contained in resource.text.
"""
title = data.get("title", "Dashboard")
theme = data.get("theme", "light")
path = uri.replace("ui://data-dashboard/", "")
if path == "metrics":
metrics = data.get("metrics", [])
if not metrics:
raise ValueError("Missing required parameter: data.metrics")
html = _build_dashboard_html(title, metrics)
return _create_ui_resource(uri, html, "100%", "auto")
elif path == "chart":
chart_type = data.get("chart_type")
chart_data = data.get("data", {})
if not chart_type:
raise ValueError("Missing required parameter: data.chart_type")
if not chart_data or not chart_data.get("series"):
raise ValueError("Missing required parameter: data.data.series")
width = data.get("width", "100%")
height = data.get("height", "400px")
stacked = data.get("stacked", False)
smooth = data.get("smooth", False)
show_label = data.get("show_label", False)
html = _build_chart_html(
title, chart_type, chart_data,
width=width, height=height, theme=theme,
stacked=stacked, smooth=smooth, show_label=show_label,
)
return _create_ui_resource(uri, html, width, height)
elif path == "multi-chart":
charts = data.get("charts", [])
if not charts:
raise ValueError("Missing required parameter: data.charts")
for i, c in enumerate(charts):
if not c.get("chart_type"):
raise ValueError(f"data.charts[{i}]: missing chart_type")
if not c.get("data", {}).get("series"):
raise ValueError(f"data.charts[{i}]: missing data.series")
columns = data.get("columns", 2)
html = _build_multi_chart_html(title, charts, columns=columns, theme=theme)
num_rows = (len(charts) + columns - 1) // columns
total_height = f"{80 + num_rows * 420}px"
return _create_ui_resource(uri, html, "100%", total_height)
else:
raise ValueError(f"Unknown URI path: {path}")
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_data_viz",
"description": "Render data visualization. Use uri to specify type.",
"inputSchema": {
"type": "object",
"properties": {
"uri": {"type": "string"},
"title": {"type": "string"},
},
"required": ["uri", "title"],
},
}
]
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_data_viz":
uri = arguments.get("uri", "")
data = arguments.get("data", {})
if not uri or not uri.startswith("ui://data-dashboard/"):
return create_error_response(
request_id, -32602,
"Invalid uri. Must be one of: ui://data-dashboard/metrics, ui://data-dashboard/chart, ui://data-dashboard/multi-chart"
)
try:
result = _handle_data_viz(uri, data)
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 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())