mcp-server/server.py

"""
MCP server — adaptive UI surface (Path B).

Tools return structuredContent that the adaptive.html widget renders
inline in MCP Apps-compatible hosts (ChatGPT, Claude, Goose, VS Code).

Run:
    python -m mcp_server.server          # stdio (local MCP client)
    python mcp-server/server.py --http   # SSE on :3100 (remote clients)
"""
import json
import os
import sys
from pathlib import Path

import httpx
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

load_dotenv()

sys.path.insert(0, str(Path(__file__).parent.parent))
from lib.router import route

HF_ENDPOINT_URL = os.getenv("HF_ENDPOINT_URL", "")
HF_TOKEN        = os.getenv("HF_TOKEN", "")
WIDGET_PATH     = Path(__file__).parent / "widget" / "adaptive.html"

mcp = FastMCP("adaptive-model", version="1.0.0")

# ── widget resource ───────────────────────────────────────────────────────────
# MCP Apps hosts load this URI into an iframe when the tool result carries
# a _widget key in structuredContent.

@mcp.resource("ui://widget/adaptive.html", mime_type="text/html")
def adaptive_widget() -> str:
    return WIDGET_PATH.read_text(encoding="utf-8")


# ── tools ─────────────────────────────────────────────────────────────────────

@mcp.tool()
async def respond(messages: list[dict], mode: str = "auto") -> dict:
    """
    Run the adaptive model for the current conversation turn.

    Args:
        messages: Full conversation history, e.g.
                  [{"role": "user", "content": "Show me revenue trends"}]
        mode:     Adapter override — 'support' | 'analytics' | 'form' | 'auto'

    Returns a dict with:
        content         — text for the model to reason over
        structuredContent — ui_spec for the widget to render interactively
    """
    resolved = route(messages, None if mode == "auto" else mode)

    headers = {"Content-Type": "application/json"}
    if HF_TOKEN:
        headers["Authorization"] = f"Bearer {HF_TOKEN}"

    async with httpx.AsyncClient(timeout=60.0) as client:
        r = await client.post(
            HF_ENDPOINT_URL,
            json={"inputs": {"messages": messages, "mode": resolved}},
            headers=headers,
        )
    r.raise_for_status()
    out = r.json()

    text    = out.get("text", "")
    ui_spec = out.get("ui_spec")

    result: dict = {
        "content": [{"type": "text", "text": text}],
    }
    if ui_spec:
        # _widget tells MCP Apps hosts which resource URI to iframe
        result["structuredContent"] = {**ui_spec, "_widget": "ui://widget/adaptive.html"}

    return result


@mcp.tool()
async def classify_mode(messages: list[dict]) -> dict:
    """
    Return which adapter would handle this turn without running inference.
    Useful for debugging routing decisions.
    """
    adapter = route(messages, None)
    return {"adapter": adapter}


# ── entry-point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--http", action="store_true",
                        help="Serve over SSE on port 3100 instead of stdio")
    args = parser.parse_args()

    if args.http:
        mcp.run(transport="sse", host="0.0.0.0", port=3100)
    else:
        mcp.run(transport="stdio")