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>
134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import streamlit as st
|
|
|
|
from .agent import Agent
|
|
from .config import load_config
|
|
from .gpfs_client import GPFSClient
|
|
from .tools import ToolDispatcher
|
|
|
|
|
|
st.set_page_config(page_title="GPFS Agent", page_icon="🗄️", layout="wide")
|
|
|
|
|
|
def _init_session() -> None:
|
|
if "cfg" in st.session_state:
|
|
return
|
|
try:
|
|
cfg = load_config()
|
|
except RuntimeError as exc:
|
|
st.error(f"Config error: {exc}")
|
|
st.stop()
|
|
|
|
st.session_state.cfg = cfg
|
|
st.session_state.client = GPFSClient(cfg)
|
|
st.session_state.tool_log: list[dict[str, Any]] = []
|
|
st.session_state.transcript: list[dict[str, str]] = []
|
|
st.session_state.allow_mutations_session = False
|
|
|
|
def confirm(message: str, call: dict[str, Any]) -> bool:
|
|
return bool(st.session_state.get("allow_mutations_session", False))
|
|
|
|
dispatcher = ToolDispatcher(st.session_state.cfg, st.session_state.client, confirm=confirm)
|
|
|
|
def on_tool(name: str, tool_input: dict[str, Any], result: dict[str, Any]) -> None:
|
|
st.session_state.tool_log.append(
|
|
{"name": name, "input": tool_input, "result": result}
|
|
)
|
|
|
|
st.session_state.agent = Agent(cfg, dispatcher, on_tool_event=on_tool)
|
|
|
|
|
|
def _mode_label(cfg) -> str:
|
|
if cfg.allow_writes and cfg.allow_destructive:
|
|
return "writes + DELETE"
|
|
if cfg.allow_writes:
|
|
return "writes (no DELETE)"
|
|
return "read-only"
|
|
|
|
|
|
def _render_sidebar() -> None:
|
|
cfg = st.session_state.cfg
|
|
with st.sidebar:
|
|
st.subheader("Session")
|
|
st.markdown(f"**Model** \n`{cfg.model}`")
|
|
st.markdown(f"**API base** \n`{cfg.gpfs_base}`")
|
|
st.markdown(f"**Mode** \n{_mode_label(cfg)}")
|
|
|
|
if cfg.allow_writes and cfg.confirm_mutations:
|
|
st.checkbox(
|
|
"Auto-approve mutations this session",
|
|
key="allow_mutations_session",
|
|
help=(
|
|
"When off, the agent's mutating calls are denied. "
|
|
"Turn this on to let it run POST/PUT" + (" / DELETE" if cfg.allow_destructive else "") + "."
|
|
),
|
|
)
|
|
elif not cfg.allow_writes:
|
|
st.caption("Writes disabled in .env (GPFS_ALLOW_WRITES=false).")
|
|
|
|
st.divider()
|
|
if st.button("Reset conversation", use_container_width=True):
|
|
st.session_state.agent.reset()
|
|
st.session_state.transcript = []
|
|
st.session_state.tool_log = []
|
|
st.rerun()
|
|
|
|
with st.expander("System prompt"):
|
|
st.code(st.session_state.agent.system_prompt, language="markdown")
|
|
|
|
with st.expander(f"Tool log ({len(st.session_state.tool_log)})"):
|
|
for entry in reversed(st.session_state.tool_log[-25:]):
|
|
ti = entry["input"]
|
|
res = entry["result"]
|
|
status = res.get("status", "ERR")
|
|
ok = res.get("ok", False)
|
|
icon = "✅" if ok else "⚠️"
|
|
st.markdown(
|
|
f"{icon} `{ti.get('method','?')} {ti.get('path','?')}` → **{status}**"
|
|
)
|
|
if ti.get("reason"):
|
|
st.caption(ti["reason"])
|
|
with st.expander("payload"):
|
|
st.code(json.dumps(res, indent=2, default=str), language="json")
|
|
|
|
|
|
def _render_transcript() -> None:
|
|
for turn in st.session_state.transcript:
|
|
with st.chat_message(turn["role"]):
|
|
st.markdown(turn["content"])
|
|
|
|
|
|
def _handle_user_input(prompt: str) -> None:
|
|
st.session_state.transcript.append({"role": "user", "content": prompt})
|
|
with st.chat_message("user"):
|
|
st.markdown(prompt)
|
|
|
|
with st.chat_message("assistant"):
|
|
with st.spinner("Thinking…"):
|
|
try:
|
|
reply = st.session_state.agent.chat(prompt)
|
|
except Exception as exc:
|
|
reply = f"_Agent error: `{exc!r}`_"
|
|
st.markdown(reply)
|
|
st.session_state.transcript.append({"role": "assistant", "content": reply})
|
|
|
|
|
|
def main() -> None:
|
|
_init_session()
|
|
_render_sidebar()
|
|
|
|
st.title("GPFS Agent")
|
|
st.caption("Chat with your IBM Spectrum Scale cluster. Guardrails come from `.env`.")
|
|
|
|
_render_transcript()
|
|
|
|
prompt = st.chat_input("Ask about the cluster…")
|
|
if prompt:
|
|
_handle_user_input(prompt)
|
|
|
|
|
|
main()
|