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>
106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Optional, Pattern
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
def _bool(name: str, default: bool = False) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None or raw == "":
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on", "y"}
|
|
|
|
|
|
def _required(name: str) -> str:
|
|
val = os.getenv(name)
|
|
if not val:
|
|
raise RuntimeError(
|
|
f"Missing required env var {name}. Copy .env.example to .env and fill it in."
|
|
)
|
|
return val
|
|
|
|
|
|
def _optional_pattern(name: str) -> Optional[Pattern[str]]:
|
|
raw = os.getenv(name)
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return re.compile(raw)
|
|
except re.error as exc:
|
|
raise RuntimeError(f"Invalid regex in {name}: {exc}") from exc
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Config:
|
|
anthropic_api_key: str
|
|
model: str
|
|
max_tokens: int
|
|
|
|
gpfs_base: str
|
|
gpfs_username: str
|
|
gpfs_password: str
|
|
verify_tls: bool | str
|
|
timeout: float
|
|
|
|
allow_writes: bool
|
|
allow_destructive: bool
|
|
confirm_mutations: bool
|
|
|
|
path_allow: Optional[Pattern[str]] = None
|
|
path_deny: Optional[Pattern[str]] = None
|
|
extra_instructions: str = ""
|
|
|
|
@property
|
|
def allowed_methods(self) -> set[str]:
|
|
methods = {"GET"}
|
|
if self.allow_writes:
|
|
methods |= {"POST", "PUT"}
|
|
if self.allow_writes and self.allow_destructive:
|
|
methods.add("DELETE")
|
|
return methods
|
|
|
|
|
|
def load_config(env_file: Optional[Path] = None) -> Config:
|
|
if env_file is not None:
|
|
load_dotenv(env_file, override=False)
|
|
else:
|
|
load_dotenv(override=False)
|
|
|
|
verify_raw = os.getenv("GPFS_VERIFY_TLS", "true").strip().lower()
|
|
ca_bundle = os.getenv("GPFS_CA_BUNDLE")
|
|
if ca_bundle:
|
|
verify: bool | str = ca_bundle
|
|
elif verify_raw in {"0", "false", "no", "off", "n"}:
|
|
verify = False
|
|
else:
|
|
verify = True
|
|
|
|
instructions = os.getenv("GPFS_INSTRUCTIONS", "").strip()
|
|
instructions_file = os.getenv("GPFS_INSTRUCTIONS_FILE")
|
|
if instructions_file:
|
|
path = Path(instructions_file)
|
|
if path.is_file():
|
|
file_text = path.read_text(encoding="utf-8").strip()
|
|
instructions = f"{instructions}\n\n{file_text}".strip() if instructions else file_text
|
|
|
|
return Config(
|
|
anthropic_api_key=_required("ANTHROPIC_API_KEY"),
|
|
model=os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6"),
|
|
max_tokens=int(os.getenv("ANTHROPIC_MAX_TOKENS", "4096")),
|
|
gpfs_base=_required("GPFS_API_BASE").rstrip("/"),
|
|
gpfs_username=_required("GPFS_USERNAME"),
|
|
gpfs_password=_required("GPFS_PASSWORD"),
|
|
verify_tls=verify,
|
|
timeout=float(os.getenv("GPFS_TIMEOUT", "30")),
|
|
allow_writes=_bool("GPFS_ALLOW_WRITES", False),
|
|
allow_destructive=_bool("GPFS_ALLOW_DESTRUCTIVE", False),
|
|
confirm_mutations=_bool("GPFS_CONFIRM_MUTATIONS", True),
|
|
path_allow=_optional_pattern("GPFS_PATH_ALLOW"),
|
|
path_deny=_optional_pattern("GPFS_PATH_DENY"),
|
|
extra_instructions=instructions,
|
|
)
|