Spaces:
Sleeping
Sleeping
| from typing import List, Dict, Tuple | |
| import os | |
| from datetime import datetime, timezone | |
| import requests | |
| from .oob_tools.utils import load_json, save_json | |
| # You can override this in HF Space secrets or env. | |
| NCS_BASE_URL = os.getenv( | |
| "NCS_BASE_URL", | |
| "https://MCP-1st-Birthday-network-change-simulator.hf.space", | |
| ) | |
| HISTORY_PATH = os.path.join( | |
| os.path.dirname(__file__), "..", "infra", "mcp_history.json" | |
| ) | |
| def _timestamp() -> str: | |
| return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") | |
| def _record_history(device: str, preset: str, risk: Dict): | |
| history = load_json(HISTORY_PATH) or {"recent": []} | |
| recent = history.get("recent", []) | |
| entry = { | |
| "timestamp": _timestamp(), | |
| "device": device, | |
| "preset": preset, | |
| "risk": risk.get("level"), | |
| "score": risk.get("score"), | |
| } | |
| recent.append(entry) | |
| # Keep history short to avoid bloat. | |
| history["recent"] = recent[-50:] | |
| save_json(HISTORY_PATH, history) | |
| def _simulate_single_step(step: Dict) -> Tuple[Dict, Dict]: | |
| """ | |
| Call the Network Change Simulator MCP-style HTTP endpoint for one step. | |
| Expects step["json"] to have: | |
| - "change_type": "vlan" | "interface" | "bgp_neighbor" | |
| - "preset_id": one of the known presets | |
| - optional "mode": "lightning" | "full" | |
| """ | |
| payload = { | |
| "change_type": step["json"]["change_type"], | |
| "preset_id": step["json"]["preset_id"], | |
| "mode": step["json"].get("mode", "lightning"), | |
| } | |
| url = f"{NCS_BASE_URL.rstrip('/')}/mcp/simulate" | |
| try: | |
| resp = requests.post(url, json=payload, timeout=10) | |
| resp.raise_for_status() | |
| except Exception as exc: # broad on purpose to keep UI resilient | |
| # Fall back to a degraded but non-crashing experience. | |
| fake = { | |
| "risk_score": 50, | |
| "risk_level": "medium", | |
| "pre_check_summary": ["Stubbed due to MCP call failure."], | |
| "post_check_summary": [], | |
| "blast_radius_summary": "MCP call failed; using placeholder blast radius.", | |
| "explanation": f"Error calling MCP server: {exc}", | |
| "mode": payload["mode"], | |
| "mode_note": "Simulated response because remote MCP call failed.", | |
| } | |
| tool_call = { | |
| "tool": "simulate_network_change", | |
| "arguments": payload, | |
| "response": fake, | |
| "error": str(exc), | |
| } | |
| risk = { | |
| "level": fake["risk_level"], | |
| "score": fake["risk_score"], | |
| "summary": fake["explanation"], | |
| } | |
| return risk, tool_call | |
| data = resp.json() | |
| # Normalize risk structure for pipeline. | |
| risk = { | |
| "level": data.get("risk_level", "unknown"), | |
| "score": data.get("risk_score", 0), | |
| "summary": data.get("explanation", ""), | |
| "blast_radius_summary": data.get("blast_radius_summary", ""), | |
| "pre_check_summary": data.get("pre_check_summary", []), | |
| "post_check_summary": data.get("post_check_summary", []), | |
| "mode": data.get("mode"), | |
| "mode_note": data.get("mode_note"), | |
| } | |
| tool_call = { | |
| "tool": "simulate_network_change", | |
| "arguments": payload, | |
| "response": data, | |
| } | |
| return risk, tool_call | |
| def simulate_steps_with_mcp(steps: List[Dict]) -> Tuple[List[Dict], List[Dict]]: | |
| """ | |
| For each step, call the Network Change Simulator MCP server. | |
| Returns: | |
| step_risks: list[dict] | |
| tool_calls: list[dict] # for logging | |
| """ | |
| step_risks: List[Dict] = [] | |
| tool_calls: List[Dict] = [] | |
| for step in steps: | |
| risk, call = _simulate_single_step(step) | |
| step_risks.append(risk) | |
| tool_calls.append(call) | |
| # Persist history for downstream root-cause inference. | |
| _record_history( | |
| device=step["json"].get("device", "unknown"), | |
| preset=step["json"].get("preset_id", "unknown"), | |
| risk=risk, | |
| ) | |
| return step_risks, tool_calls | |