Spaces:
Build error
Build error
| #!/usr/bin/env python3 | |
| """ | |
| AgentCTL Gradio UI (Frontend) | |
| Communicates with FastAPI backend instead of Kubernetes directly | |
| """ | |
| import os | |
| import time | |
| import requests | |
| import gradio as gr | |
| # ------------------------------------------------------------------- | |
| # Backend URL (FastAPI) | |
| # ------------------------------------------------------------------- | |
| BACKEND_URL = os.getenv( | |
| "AGENTCTL_BACKEND_URL", | |
| "http://127.0.0.1:9000" # default for local development | |
| ) | |
| print(f"AgentCTL Backend URL: {BACKEND_URL}") | |
| # Helper to join backend URL | |
| def api(path: str) -> str: | |
| return f"{BACKEND_URL}{path}" | |
| # ------------------------------------------------------------------- | |
| # API Client Helpers | |
| # ------------------------------------------------------------------- | |
| def generate_yaml(prompt: str, namespace: str, kind: str) -> str: | |
| """Call FastAPI: POST /api/generate-yaml""" | |
| try: | |
| payload = { | |
| "prompt": prompt, | |
| "namespace": namespace, | |
| "kind": kind | |
| } | |
| r = requests.post(api("/api/generate-yaml"), json=payload) | |
| r.raise_for_status() | |
| return r.json().get("yaml", "# No YAML returned") | |
| except Exception as e: | |
| return f"# ERROR: {e}" | |
| def apply_yaml(yaml_text: str) -> tuple[str, str]: | |
| """Call FastAPI: POST /api/apply""" | |
| try: | |
| r = requests.post(api("/api/apply"), json={"yaml": yaml_text}) | |
| r.raise_for_status() | |
| data = r.json() | |
| # Backend sends message + raw | |
| message = data.get("message", "") | |
| raw = data.get("raw", "") | |
| return message, raw | |
| except Exception as e: | |
| return f"ERROR: {e}", "" | |
| def run_agent(instruction: str, namespace: str, kind: str): | |
| """Call FastAPI: POST /api/agent""" | |
| try: | |
| payload = { | |
| "instruction": instruction, | |
| "namespace": namespace, | |
| "kind": kind | |
| } | |
| r = requests.post(api("/api/agent"), json=payload) | |
| r.raise_for_status() | |
| data = r.json() | |
| yaml_text = data.get("yaml", "# No YAML") | |
| result = data.get("result", {}) | |
| mode = data.get("mode", "unknown") | |
| model = data.get("model", "none") | |
| info = f"Mode: {mode}\nModel: {model}" | |
| return yaml_text, result, info | |
| except Exception as e: | |
| return f"# ERROR: {e}", {}, "ERROR" | |
| # ------------------------------------------------------------------- | |
| # Snapshot Formatter (Emoji-Free) | |
| # ------------------------------------------------------------------- | |
| def format_snapshot(data): | |
| lines = [] | |
| lines.append(f"Namespace: {data.get('namespace','default')}\n") | |
| lines.append("==== JOBS ====\n") | |
| for job in data.get("jobs", []): | |
| if job["succeeded"] > 0: | |
| status = "Completed" | |
| elif job["failed"] > 0: | |
| status = "Failed" | |
| elif job["active"] > 0: | |
| status = "Active" | |
| else: | |
| status = "Pending" | |
| lines.append( | |
| f"{job['name']}: {status} | Active={job['active']} | Success={job['succeeded']} | Failed={job['failed']}" | |
| ) | |
| lines.append("\n==== PODS ====\n") | |
| for pod in data.get("pods", []): | |
| phase = pod["phase"] | |
| lines.append(f"{pod['name']}: {phase}") | |
| lines.append("\n==== CRONJOBS ====\n") | |
| for cj in data.get("cronjobs", []): | |
| lines.append( | |
| f"{cj['name']}: Active={cj['active']} | LastSchedule={cj['last_schedule']}" | |
| ) | |
| return "\n".join(lines) | |
| def get_logs_once(pod_name: str, tail: int) -> str: | |
| """Call FastAPI: GET /api/logs""" | |
| try: | |
| params = {"pod_name": pod_name, "tail": tail} | |
| r = requests.get(api("/api/logs"), params=params) | |
| r.raise_for_status() | |
| return r.json().get("logs", "No logs received") | |
| except Exception as e: | |
| return f"ERROR: {e}" | |
| def follow_logs(pod_name: str, tail: int): | |
| """Streaming logs (auto-refresh every 2s)""" | |
| while True: | |
| html = get_logs_once(pod_name, tail) | |
| yield html | |
| time.sleep(2) | |
| # ------------------------------------------------------------------- | |
| # Gradio UI | |
| # ------------------------------------------------------------------- | |
| def build_ui(): | |
| with gr.Blocks(title="AgentCTL Frontend (FastAPI Backend)") as ui: | |
| gr.Markdown(f""" | |
| # AgentCTL (Frontend) | |
| Backend URL in use: | |
| **{BACKEND_URL}** | |
| Set `AGENTCTL_BACKEND_URL` to override. | |
| --- | |
| """) | |
| # ---------------------------------------------------------- | |
| # TAB 1 — Create Resource | |
| # ---------------------------------------------------------- | |
| with gr.Tab("Create Resource"): | |
| with gr.Row(): | |
| prompt = gr.Textbox( | |
| label="Describe your resource", | |
| lines=5, | |
| placeholder=( | |
| "Examples:\n" | |
| "- run a python job to preprocess data\n" | |
| "- create an nginx deployment with 3 replicas\n" | |
| "- schedule a cleanup script every 5 minutes\n" | |
| ) | |
| ) | |
| with gr.Column(): | |
| namespace = gr.Textbox(label="Namespace", value="default") | |
| kind = gr.Dropdown( | |
| label="Kind", | |
| choices=["Auto", "Job", "Deployment", "CronJob"], | |
| value="Auto" | |
| ) | |
| generate_btn = gr.Button("Generate YAML") | |
| yaml_box = gr.Code(label="Generated YAML", language="yaml") | |
| apply_btn = gr.Button("Apply YAML to Cluster") | |
| apply_status = gr.Textbox(label="Status") | |
| apply_raw = gr.Textbox(label="Raw API Response") | |
| generate_btn.click( | |
| fn=generate_yaml, | |
| inputs=[prompt, namespace, kind], | |
| outputs=yaml_box | |
| ) | |
| apply_btn.click( | |
| fn=apply_yaml, | |
| inputs=[yaml_box], | |
| outputs=[apply_status, apply_raw] | |
| ) | |
| # ---------------------------------------------------------- | |
| # TAB 2 — Cluster Dashboard | |
| # ---------------------------------------------------------- | |
| with gr.Tab("Cluster Dashboard"): | |
| refresh_button = gr.Button("Refresh Snapshot") | |
| snapshot_box = gr.Textbox(label="Cluster Snapshot", lines=25) | |
| def get_formatted_snapshot(): | |
| try: | |
| r = requests.get(api("/api/snapshot")) | |
| r.raise_for_status() | |
| data = r.json() | |
| return format_snapshot(data) | |
| except Exception as e: | |
| return f"ERROR fetching snapshot: {e}" | |
| refresh_button.click( | |
| fn=get_formatted_snapshot, | |
| outputs=snapshot_box | |
| ) | |
| # ---------------------------------------------------------- | |
| # TAB 3 — Pod Logs | |
| # ---------------------------------------------------------- | |
| with gr.Tab("Pod Logs"): | |
| pod_name = gr.Textbox(label="Pod Name") | |
| tail_lines = gr.Slider( | |
| label="Tail Lines", minimum=10, maximum=500, value=100 | |
| ) | |
| get_logs_btn = gr.Button("Get Logs") | |
| follow_logs_btn = gr.Button("Follow Logs (Live)") | |
| logs_html = gr.HTML(label="Logs Terminal") | |
| get_logs_btn.click( | |
| fn=get_logs_once, | |
| inputs=[pod_name, tail_lines], | |
| outputs=logs_html | |
| ) | |
| follow_logs_btn.click( | |
| fn=follow_logs, | |
| inputs=[pod_name, tail_lines], | |
| outputs=logs_html | |
| ) | |
| # ---------------------------------------------------------- | |
| # TAB 4 — Agent Mode (LLM → YAML → Apply) | |
| # ---------------------------------------------------------- | |
| with gr.Tab("Agent Mode"): | |
| gr.Markdown(""" | |
| ## Agentic LLM Mode | |
| Enter natural language instructions and let the backend agent: | |
| 1. Plan the Kubernetes operation | |
| 2. Generate YAML (LLM → fallback → validated) | |
| 3. Apply it to the cluster | |
| """) | |
| instruction = gr.Textbox( | |
| label="Instruction", | |
| lines=4, | |
| placeholder="e.g. deploy a pytorch training job with 1 GPU" | |
| ) | |
| namespace2 = gr.Textbox( | |
| label="Namespace", | |
| value="default" | |
| ) | |
| kind2 = gr.Dropdown( | |
| label="Kind Override (optional)", | |
| choices=["Auto", "Job", "Deployment", "CronJob"], | |
| value="Auto" | |
| ) | |
| run_btn = gr.Button("Run Agent") | |
| agent_yaml = gr.Code(label="Generated YAML", language="yaml") | |
| agent_result = gr.JSON(label="Apply Result") | |
| agent_info = gr.Textbox(label="Agent Info (LLM / Fallback)") | |
| run_btn.click( | |
| fn=run_agent, | |
| inputs=[instruction, namespace2, kind2], | |
| outputs=[agent_yaml, agent_result, agent_info] | |
| ) | |
| return ui | |
| # Build UI | |
| app = build_ui() | |
| if __name__ == "__main__": | |
| app.launch(server_name="0.0.0.0", server_port=7860) | |