267 lines
8.5 KiB
Python
267 lines
8.5 KiB
Python
#!/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())
|