532 lines
19 KiB
Python
532 lines
19 KiB
Python
#!/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'<span class="change" style="color:{color}"><span class="arrow">{arrow}</span> {_esc(change)}</span>'
|
|
|
|
cards_html += f"""
|
|
<div class="card">
|
|
<div class="card-label">{label}</div>
|
|
<div class="card-value">{value}</div>
|
|
{f'<div class="card-change">{change_html}</div>' if change_html else ''}
|
|
</div>"""
|
|
|
|
return f"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>{_esc(title)}</title>
|
|
<style>
|
|
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
|
|
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #f8fafc; padding: 24px; }}
|
|
h1 {{ font-size: 20px; font-weight: 600; color: #0f172a; margin-bottom: 20px; }}
|
|
.grid {{ display: grid; grid-template-columns: repeat(auto-fill, minmax(200px, 1fr)); gap: 16px; }}
|
|
.card {{ background: #fff; border: 1px solid #e2e8f0; border-radius: 12px; padding: 20px; transition: box-shadow 0.15s; }}
|
|
.card:hover {{ box-shadow: 0 4px 12px rgba(0,0,0,0.06); }}
|
|
.card-label {{ font-size: 13px; color: #64748b; margin-bottom: 8px; }}
|
|
.card-value {{ font-size: 28px; font-weight: 700; color: #0f172a; }}
|
|
.card-change {{ margin-top: 8px; font-size: 13px; font-weight: 500; }}
|
|
.arrow {{ font-size: 10px; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>{_esc(title)}</h1>
|
|
<div class="grid">{cards_html}
|
|
</div>
|
|
</body>
|
|
</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"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>{_esc(title)}</title>
|
|
<script src="{ECHARTS_CDN}"></script>
|
|
<style>
|
|
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
|
|
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: {bg_color}; padding: 16px; }}
|
|
#chart {{ width: {width}; height: {height}; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div id="chart"></div>
|
|
<script>
|
|
var chart = echarts.init(document.getElementById('chart'), {echarts_theme});
|
|
var option = {option_json};
|
|
chart.setOption(option);
|
|
window.addEventListener('resize', function() {{ chart.resize(); }});
|
|
</script>
|
|
</body>
|
|
</html>"""
|
|
|
|
|
|
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"""
|
|
<div class="chart-card">
|
|
<div id="{chart_id}" style="width:100%;height:{chart_height};"></div>
|
|
</div>"""
|
|
|
|
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"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>{_esc(title)}</title>
|
|
<script src="{ECHARTS_CDN}"></script>
|
|
<style>
|
|
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
|
|
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: {bg_color}; padding: 24px; color: {text_color}; }}
|
|
h1 {{ font-size: 22px; font-weight: 600; margin-bottom: 20px; text-align: center; }}
|
|
.grid {{ display: grid; grid-template-columns: repeat({columns}, 1fr); gap: 16px; }}
|
|
.chart-card {{ background: {card_bg}; border: 1px solid {border_color}; border-radius: 12px; padding: 16px; }}
|
|
@media (max-width: 768px) {{ .grid {{ grid-template-columns: 1fr; }} }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>{_esc(title)}</h1>
|
|
<div class="grid">{chart_divs}
|
|
</div>
|
|
<script>
|
|
var charts = [];{chart_scripts}
|
|
window.addEventListener('resize', function() {{ charts.forEach(function(c) {{ c.resize(); }}); }});
|
|
</script>
|
|
</body>
|
|
</html>"""
|
|
|
|
|
|
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, arguments: 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 = arguments.get("title", "Dashboard")
|
|
theme = arguments.get("theme", "light")
|
|
path = uri.replace("ui://data-dashboard/", "")
|
|
|
|
if path == "metrics":
|
|
metrics = arguments.get("metrics", [])
|
|
if not metrics:
|
|
raise ValueError("Missing required parameter: metrics")
|
|
html = _build_dashboard_html(title, metrics)
|
|
return _create_ui_resource(uri, html, "100%", "auto")
|
|
|
|
elif path == "chart":
|
|
chart_type = arguments.get("chart_type")
|
|
data = arguments.get("data", {})
|
|
if not chart_type:
|
|
raise ValueError("Missing required parameter: chart_type")
|
|
if not data or not data.get("series"):
|
|
raise ValueError("Missing required parameter: data.series")
|
|
|
|
width = arguments.get("width", "100%")
|
|
height = arguments.get("height", "400px")
|
|
stacked = arguments.get("stacked", False)
|
|
smooth = arguments.get("smooth", False)
|
|
show_label = arguments.get("show_label", False)
|
|
|
|
html = _build_chart_html(
|
|
title, chart_type, 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 = 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)
|
|
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", "")
|
|
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, 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 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())
|