qwen_agent/skills/developing/ecommerce-storefront/ecommerce_server.py

#!/usr/bin/env python3
"""
E-Commerce Storefront 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://ecommerce-storefront/product-list": "product-list.html",
    "ui://ecommerce-storefront/order-confirm": "order-confirm.html",
}

RESOURCE_DEFINITIONS = [
    {
        "uri": "ui://ecommerce-storefront/product-list",
        "name": "product-list",
        "title": "Product List",
        "description": "Interactive product cards with spec selection",
        "mimeType": RESOURCE_MIME_TYPE,
    },
    {
        "uri": "ui://ecommerce-storefront/order-confirm",
        "name": "order-confirm",
        "title": "Order Confirmation",
        "description": "Order summary with payment options",
        "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."""
    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
# ---------------------------------------------------------------------------

def _handle_render_product_list(arguments: Dict[str, Any]) -> Dict[str, Any]:
    title = arguments.get("title", "Products")
    products = arguments.get("products", [])
    if not products:
        raise ValueError("Missing required parameter: products")
    return _create_app_response(
        "ui://ecommerce-storefront/product-list",
        {"title": title, "products": products},
        "100%", "auto",
    )


def _handle_render_order_confirm(arguments: Dict[str, Any]) -> Dict[str, Any]:
    title = arguments.get("title", "Order Confirmation")
    order = arguments.get("order")
    payment = arguments.get("payment")
    if not order:
        raise ValueError("Missing required parameter: order")
    if not payment:
        raise ValueError("Missing required parameter: payment")
    if not order.get("items"):
        raise ValueError("Missing required parameter: order.items")
    return _create_app_response(
        "ui://ecommerce-storefront/order-confirm",
        {"title": title, "order": order, "payment": payment},
        "100%", "auto",
    )


TOOL_HANDLERS = {
    "render_product_list": _handle_render_product_list,
    "render_order_confirm": _handle_render_order_confirm,
}


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": "ecommerce-storefront",
                        "version": "1.0.0",
                    },
                },
            }

        elif method == "ping":
            return create_ping_response(request_id)

        elif method == "tools/list":
            tools = load_tools_from_json("ecommerce_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_product_list", "ui://ecommerce-storefront/product-list"),
                        ("render_order_confirm", "ui://ecommerce-storefront/order-confirm"),
                    ]
                ]
            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())