gpfsagent/gpfs_agent/tools.py

from __future__ import annotations

from typing import Any, Callable, Optional

from .config import Config
from .gpfs_client import GPFSClient


MUTATING_METHODS = {"POST", "PUT", "DELETE"}


def tool_schemas(cfg: Config) -> list[dict[str, Any]]:
    methods = sorted(cfg.allowed_methods)
    return [
        {
            "name": "gpfs_request",
            "description": (
                "Call the IBM Spectrum Scale (GPFS) REST API. The base URL is "
                "configured by the operator; you only supply the path after "
                "the version prefix (e.g. '/filesystems', '/cluster', "
                "'/nodes', '/filesystems/{fs}/filesets').\n\n"
                f"Allowed HTTP methods in this session: {', '.join(methods)}. "
                "Calls with other verbs will be rejected.\n\n"
                "Prefer the smallest scope when a question is specific."
            ),
            "input_schema": {
                "type": "object",
                "properties": {
                    "method": {
                        "type": "string",
                        "enum": methods,
                        "description": "HTTP verb.",
                    },
                    "path": {
                        "type": "string",
                        "description": (
                            "Path under the API base, starting with '/'. "
                            "Example: '/filesystems/gpfs0/filesets'."
                        ),
                    },
                    "query": {
                        "type": "object",
                        "description": "Optional query-string parameters.",
                        "additionalProperties": True,
                    },
                    "body": {
                        "type": "object",
                        "description": "Optional JSON request body.",
                        "additionalProperties": True,
                    },
                    "reason": {
                        "type": "string",
                        "description": (
                            "One sentence explaining why this call is needed. "
                            "Shown to the human operator for mutating calls."
                        ),
                    },
                },
                "required": ["method", "path", "reason"],
            },
        }
    ]


ConfirmFn = Callable[[str, dict[str, Any]], bool]


def _default_confirm(message: str, call: dict[str, Any]) -> bool:
    print(f"\n[confirm] {message}")
    print(f"  {call['method']} {call['path']}")
    if call.get("body"):
        print(f"  body: {call['body']}")
    if call.get("reason"):
        print(f"  reason: {call['reason']}")
    answer = input("  proceed? [y/N] ").strip().lower()
    return answer in {"y", "yes"}


class ToolDispatcher:
    def __init__(
        self,
        cfg: Config,
        client: GPFSClient,
        confirm: Optional[ConfirmFn] = None,
    ) -> None:
        self._cfg = cfg
        self._client = client
        self._confirm = confirm or _default_confirm

    def dispatch(self, name: str, tool_input: dict[str, Any]) -> dict[str, Any]:
        if name != "gpfs_request":
            return {"error": f"Unknown tool: {name}"}

        method = str(tool_input.get("method", "")).upper()
        path = str(tool_input.get("path", ""))
        query = tool_input.get("query") or None
        body = tool_input.get("body") or None
        reason = str(tool_input.get("reason", "")).strip()

        if not method or not path:
            return {"error": "Both 'method' and 'path' are required."}

        if method not in self._cfg.allowed_methods:
            return {
                "error": (
                    f"Method {method} is not allowed in this session. "
                    f"Allowed: {sorted(self._cfg.allowed_methods)}."
                )
            }

        if not path.startswith("/"):
            path = "/" + path

        if self._cfg.path_allow and not self._cfg.path_allow.search(path):
            return {"error": f"Path {path!r} is not in the configured allow-list."}
        if self._cfg.path_deny and self._cfg.path_deny.search(path):
            return {"error": f"Path {path!r} matches the configured deny-list."}

        if method in MUTATING_METHODS and self._cfg.confirm_mutations:
            ok = self._confirm(
                f"Claude wants to perform a mutating {method} call.",
                {"method": method, "path": path, "body": body, "reason": reason},
            )
            if not ok:
                return {
                    "error": "Operator denied this call. Do not retry without "
                    "asking the user."
                }

        try:
            result = self._client.request(method, path, params=query, body=body)
        except Exception as exc:
            return {"error": f"Transport error: {exc!r}"}

        return result