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 go through the unified merge surface in :mod:`api.skill_catalog_view`
(synced + local, with override precedence); 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).

Authoring tools (``create_skill``/``update_skill``/``delete_skill``) dispatch by
origin (design D10): they edit/delete local skills and create/update/remove
local overrides for cloud skills, never writing to the synced store.
"""

from __future__ import annotations

from collections.abc import Callable
from typing import Any

from skills_learning.models import (
    FlowStep,
    FlowTemplateSkill,
    KnowledgeSkill,
    SkillMetadata,
)
from api.skill_catalog_view import SkillCatalogView, SkillSummary, SkillView
from storage.local_skills import LocalSkillStore
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",
    "create_skill",
    "update_skill",
    "delete_skill",
)


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."""


class SkillAuthoringError(SkillCatalogError):
    """Raised when an authoring operation cannot be applied (e.g., deleting a
    cloud skill that has no local override)."""


def skill_tool_handlers(
    *,
    store: SkillCatalogStore,
    local_store: LocalSkillStore,
    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`).
    """
    view = SkillCatalogView(store, local_store)
    local = local_store
    tools_getter = get_registered_tools or (lambda: set())

    def _list_skills() -> dict[str, Any]:
        summaries = view.list_skills(get_active_subscriptions())
        return {"ok": True, "skills": [_summary_to_dict(s) for s in summaries]}

    def _search_skills(query: str) -> dict[str, Any]:
        summaries = view.search_skills(query, get_active_subscriptions())
        return {"ok": True, "skills": [_summary_to_dict(s) for s in summaries]}

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

    def _resolve_flow_template(
        skill_id: str,
        params: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        params = params or {}
        result = view.get_skill(
            skill_id,
            get_active_subscriptions(),
            registered_tools=tools_getter(),
        )
        if result is None:
            return _error_response(SkillNotFoundError(skill_id))
        skill = result.skill
        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}

    def _create_skill(payload: dict[str, Any]) -> dict[str, Any]:
        try:
            skill = _build_skill(payload)
        except ValueError as exc:
            return _error_response(SkillAuthoringError(str(exc)))
        stored = local.create_local(skill)
        return {"ok": True, "skill": _skill_to_full_dict(stored), "origin": "local"}

    def _update_skill(skill_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        try:
            skill = _build_skill(payload)
        except ValueError as exc:
            return _error_response(SkillAuthoringError(str(exc)))
        if view.is_local_skill(skill_id):
            stored = local.update_local(_with_id(skill, skill_id))
            return {
                "ok": True,
                "skill": _skill_to_full_dict(stored),
                "origin": "local",
            }
        # Cloud skill id (or anticipated one): create/update a local override.
        stored = local.upsert_override(skill_id, skill)
        return {
            "ok": True,
            "skill": _skill_to_full_dict(stored),
            "origin": "cloud",
            "locally_overridden": True,
        }

    def _delete_skill(skill_id: str) -> dict[str, Any]:
        if view.is_local_skill(skill_id):
            local.delete_local(skill_id)
            return {"ok": True, "deleted": skill_id, "origin": "local"}
        if view.has_override(skill_id):
            local.remove_override(skill_id)
            return {
                "ok": True,
                "deleted_override": skill_id,
                "origin": "cloud",
            }
        return _error_response(
            SkillAuthoringError(
                f"skill {skill_id} is a cloud skill with no local override to remove"
            )
        )

    return {
        "list_skills": _list_skills,
        "search_skills": _search_skills,
        "get_skill": _get_skill,
        "resolve_flow_template": _resolve_flow_template,
        "create_skill": _create_skill,
        "update_skill": _update_skill,
        "delete_skill": _delete_skill,
    }


def register_skill_catalog_tools(
    server: Any,
    *,
    store: SkillCatalogStore,
    local_store: LocalSkillStore,
    get_active_subscriptions: Callable[[], set[str]],
    get_registered_tools: Callable[[], set[str]] | None = None,
) -> Any:
    """Register the skill MCP tools (read + authoring) 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. Authoring tools
    are always registered (design D6): there is no enable/disable gate.
    """
    handlers = skill_tool_handlers(
        store=store,
        local_store=local_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)

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

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

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

    return server


# ----------------------------------------------------------------------
# Serialization helpers
# ----------------------------------------------------------------------


def _summary_to_dict(summary: SkillSummary) -> dict[str, Any]:
    meta = summary.metadata
    return {
        "id": meta.id,
        "name": meta.name,
        "description": meta.description,
        "kind": meta.kind,
        "tags": list(meta.tags),
        "origin": summary.origin,
        "locally_overridden": summary.locally_overridden,
    }


def _view_to_dict(view: SkillView) -> dict[str, Any]:
    base = _skill_to_full_dict(view.skill)
    base["origin"] = view.origin
    base["locally_overridden"] = view.locally_overridden
    return base


def _skill_to_full_dict(skill: Any) -> dict[str, Any]:
    """Full skill payload for ``get_skill`` / authoring responses."""
    meta: SkillMetadata = skill.metadata
    base = {
        "id": meta.id,
        "name": meta.name,
        "description": meta.description,
        "kind": meta.kind,
        "tags": list(meta.tags),
        "version": meta.version,
        "updated_at": meta.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 _build_skill(payload: dict[str, Any]) -> KnowledgeSkill | FlowTemplateSkill:
    """Construct a Skill from an authoring payload."""
    kind = str(payload.get("kind") or "").strip()
    name = str(payload.get("name") or "").strip()
    if not name:
        raise ValueError("skill name must not be empty")
    if kind not in ("knowledge", "flow_template"):
        raise ValueError(f"unsupported skill kind: {kind!r}")
    tags = [str(tag) for tag in payload.get("tags") or []]
    meta = SkillMetadata(name=name, kind=kind, tags=tags)  # type: ignore[arg-type]
    if kind == "knowledge":
        return KnowledgeSkill(metadata=meta, content=str(payload.get("content") or ""))
    steps = [
        FlowStep(
            tool_name=str(step.get("tool_name") or step.get("action") or ""),
            args=dict(step.get("args") or {}),
        )
        for step in (payload.get("steps") or [])
    ]
    parameters = {
        str(name): dict(schema)
        for name, schema in (payload.get("parameters") or {}).items()
    }
    return FlowTemplateSkill(metadata=meta, steps=steps, parameters=parameters)


def _with_id(skill: KnowledgeSkill | FlowTemplateSkill, skill_id: str):
    meta = skill.metadata
    from dataclasses import replace

    new_meta = replace(meta, id=skill_id)
    if isinstance(skill, KnowledgeSkill):
        return KnowledgeSkill(metadata=new_meta, content=skill.content)
    return FlowTemplateSkill(
        metadata=new_meta,
        steps=list(skill.steps),
        parameters={n: dict(s) for n, s in skill.parameters.items()},
    )


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"
    if isinstance(exc, SkillAuthoringError):
        message = str(exc)
        return f"authoring error: {message}" if message else "authoring error"
    return "operation failed"