""" REST endpoint for serving MCP App resources. Maps ui:// URIs to static HTML App files under the bot's skill directory. GET /api/v1/mcp/resources?uri=ui://data-dashboard/chart&bot_id=xxx -> projects/robot/{bot_id}/skills/data-dashboard/apps/chart.html """ import logging from pathlib import Path from fastapi import APIRouter, Query, HTTPException from fastapi.responses import HTMLResponse logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/mcp", tags=["mcp-resources"]) PROJECT_ROOT = Path(__file__).parent.parent # In-memory cache: (bot_id, uri) -> html string _resource_cache: dict[tuple[str, str], str] = {} def _resolve_uri_to_path(uri: str, bot_id: str) -> Path: """Resolve a ui:// URI to a local file path. URI format: ui://{server-name}/{resource-name} File path: projects/robot/{bot_id}/skills/{server-name}/apps/{resource-name}.html """ if not uri.startswith("ui://"): raise ValueError(f"Invalid URI scheme: {uri}") rest = uri[5:] # Remove "ui://" parts = rest.split("/", 1) if len(parts) != 2 or not parts[0] or not parts[1]: raise ValueError(f"Invalid URI format: {uri}") server_name, resource_name = parts # Prevent path traversal if ".." in server_name or "/" in server_name: raise ValueError(f"Invalid server name: {server_name}") safe_name = Path(resource_name).name if safe_name != resource_name: raise ValueError(f"Invalid resource name: {resource_name}") return PROJECT_ROOT / "projects" / "robot" / bot_id / "skills" / server_name / "apps" / f"{resource_name}.html" @router.get("/resources") async def get_resource( uri: str = Query(..., description="Resource URI (e.g. ui://data-dashboard/chart)"), bot_id: str = Query(..., description="Bot ID"), ): """Fetch an MCP App HTML resource by URI and bot_id. Returns the static HTML App that should be loaded into an iframe. The host sends tool result data to the iframe via postMessage. """ cache_key = (bot_id, uri) if cache_key in _resource_cache: return HTMLResponse(content=_resource_cache[cache_key], media_type="text/html") try: file_path = _resolve_uri_to_path(uri, bot_id) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) if not file_path.is_file(): raise HTTPException(status_code=404, detail=f"Resource not found: {uri} (bot_id={bot_id})") html = file_path.read_text(encoding="utf-8") _resource_cache[cache_key] = html return HTMLResponse(content=html, media_type="text/html")