Initial commit: agentic chat app for IBM Storage Scale REST API v3

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>
This commit is contained in:
Laurence Horrocks-Barlow 2026-05-20 01:36:59 +01:00
commit 8840ba74f7
13 changed files with 959 additions and 0 deletions

74
gpfs_agent/gpfs_client.py Normal file
View file

@ -0,0 +1,74 @@
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,
}