Laurence Horrocks-Barlow
8840ba74f7
Adds a Python package that wraps the Storage Scale /scalemgmt/v3 management API behind a Claude-driven tool-use loop. Ships a Rich-styled terminal REPL and a Streamlit web UI sharing the same Agent/Dispatcher/GPFSClient stack. Guardrails are env-driven (.env): read-only by default; writes and DELETE gated by GPFS_ALLOW_WRITES / GPFS_ALLOW_DESTRUCTIVE; optional path allow/deny regex; mutating calls require operator confirmation. Free-text GPFS_INSTRUCTIONS (or GPFS_INSTRUCTIONS_FILE) appended to the system prompt. The system prompt also includes a curated v3 endpoint reference compiled from IBM Storage Scale 5.2.3 / 6.0.0 documentation so the agent doesn't have to guess endpoint shapes for common operations. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
74 lines
1.9 KiB
Python
74 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Optional
|
|
|
|
import httpx
|
|
|
|
from .config import Config
|
|
|
|
|
|
class GPFSError(RuntimeError):
|
|
def __init__(self, status: int, body: Any, method: str, url: str) -> None:
|
|
super().__init__(f"{method} {url} -> HTTP {status}")
|
|
self.status = status
|
|
self.body = body
|
|
self.method = method
|
|
self.url = url
|
|
|
|
|
|
class GPFSClient:
|
|
def __init__(self, cfg: Config) -> None:
|
|
self._cfg = cfg
|
|
self._client = httpx.Client(
|
|
base_url=cfg.gpfs_base,
|
|
auth=(cfg.gpfs_username, cfg.gpfs_password),
|
|
verify=cfg.verify_tls,
|
|
timeout=cfg.timeout,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
|
|
def close(self) -> None:
|
|
self._client.close()
|
|
|
|
def __enter__(self) -> "GPFSClient":
|
|
return self
|
|
|
|
def __exit__(self, *_: object) -> None:
|
|
self.close()
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
params: Optional[dict[str, Any]] = None,
|
|
body: Optional[dict[str, Any]] = None,
|
|
) -> dict[str, Any]:
|
|
if not path.startswith("/"):
|
|
path = "/" + path
|
|
|
|
response = self._client.request(
|
|
method=method.upper(),
|
|
url=path,
|
|
params=params or None,
|
|
content=json.dumps(body) if body is not None else None,
|
|
)
|
|
|
|
text = response.text
|
|
data: Any
|
|
try:
|
|
data = response.json() if text else None
|
|
except json.JSONDecodeError:
|
|
data = None
|
|
|
|
return {
|
|
"status": response.status_code,
|
|
"ok": response.is_success,
|
|
"url": str(response.request.url),
|
|
"method": method.upper(),
|
|
"data": data,
|
|
"text": None if data is not None else text,
|
|
}
|