agentic-mobile-control/api/skill_catalog_mcp.py

"""MCP tool surface for the Skill Catalog.

API layer per CONSTITUTION.md: MCP dependencies (FastMCP) live here.
Reads from :mod:`storage.skill_catalog`; flow-template parameter resolution
delegates to the existing :func:`workflow.skill_exec.resolve_skill_steps`.
No server-side execution primitive is exposed — flow templates are resolved
here but executed step-by-step by the LLM via the existing device-capability
tools (design D5).
"""
from __future__ import annotations

from collections.abc import Callable
from typing import Any

from skills_learning.models import (
    FlowTemplateSkill,
    KnowledgeSkill,
    SkillMetadata,
)
from storage.skill_catalog import SkillCatalogStore
from workflow.skill_exec import SkillExecutionError, resolve_skill_steps

# Tool names registered by this module. Used by tests and by callers that
# need to assert the full registered set (e.g., no batch-execute tool).
SKILL_TOOL_NAMES = (
    "list_skills",
    "search_skills",
    "get_skill",
    "resolve_flow_template",
)


class SkillCatalogError(Exception):
    """Base for skill MCP semantic errors."""


class SkillNotFoundError(SkillCatalogError):
    """Raised when a skill id is unknown or not visible to the caller.

    Both cases raise the same error to avoid leaking existence (design D4,
    task 2.4 indistinguishability contract).
    """


class InvalidFlowTemplateError(SkillCatalogError):
    """Raised when a skill exists but cannot be returned as a flow template
    (e.g., it's a knowledge skill, or a step references an unknown tool)."""


class MissingParameterError(SkillCatalogError):
    """Raised when required flow-template parameters are missing/invalid."""


def skill_tool_handlers(
    *,
    store: SkillCatalogStore,
    get_active_subscriptions: Callable[[], set[str]],
    get_registered_tools: Callable[[], set[str]] | None = None,
) -> dict[str, Callable[..., dict[str, Any]]]:
    """Return a dict of MCP tool handler functions, keyed by tool name.

    Decoupled from FastMCP so handlers can be tested directly without
    standing up a server (mirrors :func:`api.mcp.tool_handlers`).
    """
    tools_getter = get_registered_tools or (lambda: set())

    def _list_skills() -> dict[str, Any]:
        metas = store.list_skills(get_active_subscriptions())
        return {
            "ok": True,
            "skills": [_metadata_to_summary(m) for m in metas],
        }

    def _search_skills(query: str) -> dict[str, Any]:
        metas = store.search_skills(query, get_active_subscriptions())
        return {
            "ok": True,
            "skills": [_metadata_to_summary(m) for m in metas],
        }

    def _get_skill(skill_id: str) -> dict[str, Any]:
        skill = store.get_skill(
            skill_id,
            get_active_subscriptions(),
            registered_tools=tools_getter(),
        )
        if skill is None:
            return _error_response(SkillNotFoundError(skill_id))
        return {"ok": True, "skill": _skill_to_full_dict(skill)}

    def _resolve_flow_template(
        skill_id: str,
        params: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        params = params or {}
        skill = store.get_skill(
            skill_id,
            get_active_subscriptions(),
            registered_tools=tools_getter(),
        )
        if skill is None:
            return _error_response(SkillNotFoundError(skill_id))
        if not isinstance(skill, FlowTemplateSkill):
            return _error_response(
                InvalidFlowTemplateError(
                    f"skill {skill_id} is not a flow_template (kind={skill.metadata.kind})"
                )
            )
        try:
            steps = resolve_skill_steps(skill, params)
        except SkillExecutionError as exc:
            return _error_response(MissingParameterError(str(exc)))
        return {"ok": True, "steps": steps}

    return {
        "list_skills": _list_skills,
        "search_skills": _search_skills,
        "get_skill": _get_skill,
        "resolve_flow_template": _resolve_flow_template,
    }


def register_skill_catalog_tools(
    server: Any,
    *,
    store: SkillCatalogStore,
    get_active_subscriptions: Callable[[], set[str]],
    get_registered_tools: Callable[[], set[str]] | None = None,
) -> Any:
    """Register ``list_skills``/``search_skills``/``get_skill``/
    ``resolve_flow_template`` as MCP tools on ``server``.

    Returns the server so the caller can chain. No batch-execute tool is
    registered (design D5): the LLM issues each resulting device-capability
    tool call itself, preserving the Observe-Think-Act loop.
    """
    handlers = skill_tool_handlers(
        store=store,
        get_active_subscriptions=get_active_subscriptions,
        get_registered_tools=get_registered_tools,
    )

    @server.tool(name="list_skills")
    def _list_skills() -> dict[str, Any]:
        return handlers["list_skills"]()

    @server.tool(name="search_skills")
    def _search_skills(query: str) -> dict[str, Any]:
        return handlers["search_skills"](query=query)

    @server.tool(name="get_skill")
    def _get_skill(skill_id: str) -> dict[str, Any]:
        return handlers["get_skill"](skill_id=skill_id)

    @server.tool(name="resolve_flow_template")
    def _resolve_flow_template(
        skill_id: str,
        params: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        return handlers["resolve_flow_template"](skill_id=skill_id, params=params)

    return server


def _metadata_to_summary(meta: SkillMetadata) -> dict[str, Any]:
    """Compact metadata for list/search — no Subscription-Platform-specific fields."""
    return {
        "id": meta.id,
        "name": meta.name,
        "description": meta.description,
        "kind": meta.kind,
        "tags": list(meta.tags),
    }


def _skill_to_full_dict(skill: Any) -> dict[str, Any]:
    """Full skill payload for ``get_skill``."""
    base = _metadata_to_summary(skill.metadata)
    base["version"] = skill.metadata.version
    base["updated_at"] = skill.metadata.updated_at.isoformat()
    if isinstance(skill, KnowledgeSkill):
        base["content"] = skill.content
    elif isinstance(skill, FlowTemplateSkill):
        base["steps"] = [step.to_dict() for step in skill.steps]
        base["parameters"] = {
            name: dict(schema) for name, schema in skill.parameters.items()
        }
    return base


def _error_response(exc: Exception) -> dict[str, Any]:
    return {"ok": False, "error": _semantic_skill_error(exc)}


def _semantic_skill_error(exc: Exception) -> str:
    if isinstance(exc, SkillNotFoundError):
        return "skill not found"
    if isinstance(exc, InvalidFlowTemplateError):
        return "skill unavailable"
    if isinstance(exc, MissingParameterError):
        message = str(exc)
        return f"missing parameter: {message}" if message else "invalid parameter"
    return "operation failed"