78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""
|
|
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")
|