Host-owned routing for direct and MCP execution

# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Sarang Datye
"""Host-owned routing from validated domain intent to approved execution."""

from __future__ import annotations

import json
from collections.abc import Callable
from typing import Any, Protocol

from mcp.types import CallToolResult, ListToolsResult

from deterministic_capabilities import (
    calculate,
    create_support_ticket,
    validate_email,
    validate_request,
)


class MCPExecutionService(Protocol):
    async def list_tools(self) -> ListToolsResult:
        """Return the server's advertised tools."""

    async def call_tool(self, name: str, arguments: dict[str, Any]) -> CallToolResult:
        """Invoke an MCP tool."""


ToolHandler = Callable[..., dict[str, Any]]

DIRECT_CAPABILITIES: dict[str, ToolHandler] = {
    "calculator": calculate,
    "email_validator": validate_email,
    "create_support_ticket": create_support_ticket,
}

MCP_CAPABILITIES = {
    "calculator": "calculate",
    "email_validator": "validate_email",
    "create_support_ticket": "create_support_ticket",
}

IMPLEMENTATIONS = {
    "calculator": "deterministic_capabilities.py",
    "email_validator": "deterministic_capabilities.py",
    "create_support_ticket": "deterministic_capabilities.py",
}


def required_mcp_tools() -> frozenset[str]:
    return frozenset(MCP_CAPABILITIES.values())


async def require_approved_mcp_tools(service: MCPExecutionService) -> None:
    """Discover server tools once and require every approved tool to be present."""
    listing = await service.list_tools()
    available = {tool.name for tool in listing.tools}
    missing = required_mcp_tools() - available
    if missing:
        names = ", ".join(sorted(missing))
        raise ValueError(f"MCP server is missing required tool(s): {names}.")


def normalize_call_tool_result(result: CallToolResult) -> dict[str, Any]:
    """Convert MCP CallToolResult into the dict shape the host expects."""
    if result.is_error:
        raise RuntimeError("MCP tool returned an error result.")

    structured = getattr(result, "structured_content", None)
    if structured is None:
        structured = getattr(result, "structuredContent", None)
    if structured is not None:
        if not isinstance(structured, dict):
            raise TypeError("MCP structured content must be a JSON object.")
        return structured

    if len(result.content) != 1 or getattr(result.content[0], "type", None) != "text":
        raise TypeError("MCP tool result must have structured content or one JSON text block.")

    parsed = json.loads(result.content[0].text)
    if not isinstance(parsed, dict):
        raise TypeError("MCP JSON text content must decode to an object.")
    return parsed


def execute_direct(request: dict[str, Any]) -> dict[str, Any]:
    """Validate the request, choose an approved function, and call normal Python."""
    validated = validate_request(request)
    handler = DIRECT_CAPABILITIES.get(validated.intent)
    if handler is None:
        raise ValueError(f"Intent '{validated.intent}' is not approved for direct execution.")
    return {
        "intent": validated.intent,
        "implementation": IMPLEMENTATIONS[validated.intent],
        "output": handler(**validated.parameters),
    }


async def execute_via_mcp(request: dict[str, Any], service: MCPExecutionService) -> dict[str, Any]:
    """Validate the request, choose an approved MCP tool name, and invoke MCP."""
    validated = validate_request(request)
    tool_name = MCP_CAPABILITIES.get(validated.intent)
    if tool_name is None:
        raise ValueError(f"Intent '{validated.intent}' is not approved for MCP execution.")
    await require_approved_mcp_tools(service)
    result = await service.call_tool(tool_name, dict(validated.parameters))
    output = normalize_call_tool_result(result)
    return {
        "intent": validated.intent,
        "mcp_tool": tool_name,
        "transport": "provided-by-client",
        "implementation": IMPLEMENTATIONS[validated.intent],
        "llm_calls_for_execution": 0,
        "output": output,
    }

python3 -m py_compile deterministic_capabilities.py host_authorized_routing.py