python-mcp/tools/info.py

from __future__ import annotations

from typing import Any
import inspect

from core.tools.base import BaseTool, ToolContext
from core.tools.registry import registry
from core.events import bus


class InfoTool(BaseTool):
    """
    Introspective tool for:
    - tool discovery
    - execution signature inspection
    - structured payload hints
    - example generation
    """

    name = "info"
    description = "Get tool schemas, args, and usage hints"

    # =========================================================
    # EXECUTE
    # =========================================================

    def execute(self, payload: dict[str, Any], ctx: ToolContext):
        action = payload.get("action", "list_tools")

        bus.log(
            "INFO",
            "info_execute",
            "INFO",
            {"action": action}
        )

        match action:
            case "list_tools":
                return self.list_tools()

            case "tool_schema":
                return self.tool_schema(payload)

            case "tool_catalog":
                return self.tool_catalog()

            case _:
                raise ValueError(f"Unknown info action: {action}")

    # =========================================================
    # LIST TOOLS
    # =========================================================

    def list_tools(self) -> dict[str, Any]:
        return {
            "tools": [
                {
                    "name": t.name,
                    "description": getattr(t, "description", "")
                }
                for t in registry.all_tools()
            ]
        }

    # =========================================================
    # TOOL SCHEMA (CORE FEATURE)
    # =========================================================

    def tool_schema(self, payload: dict[str, Any]) -> dict[str, Any]:
        name = payload.get("name")

        if not isinstance(name, str):
            raise ValueError("name must be string")

        tool = next((t for t in registry.all_tools() if t.name == name), None)

        if not tool:
            return {"error": f"Tool not found: {name}"}

        schema: dict[str, Any] = {
            "name": tool.name,
            "description": getattr(tool, "description", ""),
            "class": tool.__class__.__name__,
            "module": tool.__class__.__module__,
        }

        # =====================================================
        # SAFE SIGNATURE INTROSPECTION
        # =====================================================
        try:
            sig = inspect.signature(tool.execute)

            params: dict[str, Any] = {}

            for pname, param in sig.parameters.items():
                if pname == "self":
                    continue

                params[pname] = {
                    "required": param.default is inspect._empty,
                    "default": None if param.default is inspect._empty else param.default,
                    "kind": str(param.kind),
                    "type": (
                        param.annotation.__name__
                        if hasattr(param.annotation, "__name__")
                        else str(param.annotation)
                    )
                }

            schema["execute_signature"] = params

        except Exception as e:
            schema["execute_signature_error"] = str(e)

        # =====================================================
        # STRUCTURED HINTS (NOT STRING PARSING)
        # =====================================================
        schema["common_payload_patterns"] = self._infer_patterns(tool)
        schema["example_payload"] = self._generate_example_payload(tool)

        return schema

    # =========================================================
    # TOOL CATALOG
    # =========================================================

    def tool_catalog(self) -> dict[str, Any]:
        return {
            "tool_count": len(registry.all_tools()),
            "tools": [
                {
                    "name": t.name,
                    "description": getattr(t, "description", ""),
                    "patterns": self._infer_patterns(t),
                }
                for t in registry.all_tools()
            ]
        }

    # =========================================================
    # PATTERN INFERENCE (CLEANED UP)
    # =========================================================

    def _infer_patterns(self, tool: BaseTool) -> list[str]:
        name = tool.name.lower()

        mapping = {
            "git": ["repo", "branch", "remote", "message"],
            "filesystem": ["path", "content", "directory"],
            "gitea": ["owner", "repo", "path", "message"],
            "subprocess": ["cmd", "cwd", "timeout"],
            "memory": ["entry", "query"],
            "reflection": ["action"],
        }

        return mapping.get(name, [])

    # =========================================================
    # EXAMPLES
    # =========================================================

    def _generate_example_payload(self, tool: BaseTool) -> dict[str, Any]:
        name = tool.name.lower()

        examples = {
            "git": {
                "action": "status",
                "repo": "."
            },
            "filesystem": {
                "action": "read_file",
                "path": "./example.txt"
            },
            "gitea": {
                "action": "list_repos"
            },
            "subprocess": {
                "cmd": ["echo", "hello"],
                "timeout": 60
            },
            "memory": {
                "action": "add",
                "entry": {"note": "example"}
            },
            "reflection": {
                "action": "reflect"
            }
        }

        return examples.get(name, {"action": "unknown"})