Spaces:
Sleeping
Sleeping
| """ | |
| Wireless Core Services AI Agent Flow - Hugging Face / Gradio Demo | |
| A runnable synthetic telecom operations demo emphasizing: | |
| - Supervisor agent model | |
| - MCP-style tool registry and data connector layer | |
| - Guardrailed mitigation recommendations | |
| - Synthetic data generation for testing | |
| This demo is intentionally offline and self-contained. It simulates Claude/Copilot-style | |
| reasoning platforms with deterministic Python logic so it can run in Hugging Face Spaces | |
| without API keys. Replace the `SupervisorAgent.reason()` method or individual MCP tool | |
| functions with real model/tool calls when connecting to enterprise systems. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import math | |
| import random | |
| import textwrap | |
| import uuid | |
| from dataclasses import dataclass, field | |
| from datetime import datetime, timedelta | |
| from typing import Any, Callable, Dict, List, Optional, Tuple | |
| import gradio as gr | |
| import numpy as np | |
| import pandas as pd | |
| import plotly.graph_objects as go | |
| APP_TITLE = "Wireless Core Services AI Agent Flow" | |
| APP_SUBTITLE = "Supervisor model + MCP tooling + synthetic telecom operations data" | |
| DOMAINS = [ | |
| "Packet Core", | |
| "Voice Core", | |
| "Messaging", | |
| "E911", | |
| "USD", | |
| "Policy & Charging", | |
| ] | |
| PLATFORMS = { | |
| "Packet Core": ["AMF", "SMF", "UPF", "AUSF", "CHF"], | |
| "Voice Core": ["IMS", "SBC", "TAS", "HSS"], | |
| "Messaging": ["SMSC", "MMSC", "IP-SM-GW"], | |
| "E911": ["E911 Router", "ALI DB", "PSAP Gateway"], | |
| "USD": ["USD Gateway", "USSD Session Manager"], | |
| "Policy & Charging": ["PCF", "PCRF", "OCS", "Charging Gateway"], | |
| } | |
| SCENARIOS = { | |
| "AMF Overload Detected": { | |
| "domain": "Packet Core", | |
| "platform": "AMF", | |
| "symptoms": ["registration failures", "high CPU", "attach latency", "alarm burst"], | |
| "fault_type": "Capacity / control-plane overload", | |
| "recommended_action": "Scale out AMF instances in us-east-1 and shift 10% registration load to standby pool.", | |
| "risk_hint": "low", | |
| "expected_root_cause": "Control-plane load increased after a regional reconnect storm.", | |
| }, | |
| "SMF Latency Elevated": { | |
| "domain": "Packet Core", | |
| "platform": "SMF", | |
| "symptoms": ["PDU session setup latency", "UPF selection delay", "ticket cluster"], | |
| "fault_type": "Session management latency", | |
| "recommended_action": "Drain new PDU sessions from the degraded SMF pool and validate UPF selection latency.", | |
| "risk_hint": "medium", | |
| "expected_root_cause": "A policy lookup slowdown is delaying session establishment.", | |
| }, | |
| "PCF Policy Failures": { | |
| "domain": "Policy & Charging", | |
| "platform": "PCF", | |
| "symptoms": ["policy decision failures", "charging API errors", "recent config change"], | |
| "fault_type": "Policy rule regression", | |
| "recommended_action": "Rollback the latest PCF rule bundle after human approval and validate charging calls.", | |
| "risk_hint": "high", | |
| "expected_root_cause": "A recent rule deployment is returning invalid policy decisions for a subscriber segment.", | |
| }, | |
| "E911 Routing Anomaly": { | |
| "domain": "E911", | |
| "platform": "E911 Router", | |
| "symptoms": ["emergency route mismatch", "PSAP handoff delay", "critical alarm"], | |
| "fault_type": "Emergency routing risk", | |
| "recommended_action": "Escalate to E911 operations; do not auto-execute routing changes. Validate PSAP route table manually.", | |
| "risk_hint": "critical", | |
| "expected_root_cause": "Emergency routing table and PSAP handoff data are inconsistent.", | |
| }, | |
| "IMS Voice Degradation": { | |
| "domain": "Voice Core", | |
| "platform": "IMS", | |
| "symptoms": ["call setup failures", "SIP 5xx", "SBC adjacency errors"], | |
| "fault_type": "Voice call setup degradation", | |
| "recommended_action": "Route a limited percentage of IMS traffic through the healthy SBC pair and monitor SIP error recovery.", | |
| "risk_hint": "medium", | |
| "expected_root_cause": "IMS call setup failures correlate with SBC adjacency errors.", | |
| }, | |
| "Kafka Telemetry Backlog": { | |
| "domain": "Packet Core", | |
| "platform": "Kafka Telemetry Pipeline", | |
| "symptoms": ["delayed KPI ingestion", "lag spike", "missing near-real-time alerts"], | |
| "fault_type": "Observability pipeline backlog", | |
| "recommended_action": "Scale Kafka consumers and temporarily reduce non-critical enrichment jobs.", | |
| "risk_hint": "low", | |
| "expected_root_cause": "Consumer lag is delaying telemetry used by fault detection agents.", | |
| }, | |
| } | |
| # ----------------------------------------------------------------------------- | |
| # Synthetic data generation | |
| # ----------------------------------------------------------------------------- | |
| class SyntheticTelecomDataset: | |
| seed: int = 42 | |
| minutes: int = 120 | |
| scenario_name: str = "AMF Overload Detected" | |
| generated_at: datetime = field(default_factory=datetime.utcnow) | |
| def __post_init__(self) -> None: | |
| self.rng = random.Random(self.seed) | |
| self.np_rng = np.random.default_rng(self.seed) | |
| self.scenario = SCENARIOS[self.scenario_name] | |
| self.kpi_df = self._generate_kpis() | |
| self.alerts_df = self._generate_alerts() | |
| self.tickets_df = self._generate_tickets() | |
| self.changes_df = self._generate_changes() | |
| self.topology_df = self._generate_topology() | |
| self.runbooks = self._generate_runbooks() | |
| self.audit_df = pd.DataFrame( | |
| columns=["timestamp", "actor", "event", "status", "detail"] | |
| ) | |
| def _platform_rows(self) -> List[Tuple[str, str]]: | |
| rows: List[Tuple[str, str]] = [] | |
| for domain, platforms in PLATFORMS.items(): | |
| for platform in platforms: | |
| rows.append((domain, platform)) | |
| rows.append(("Data Processing", "Kafka Telemetry Pipeline")) | |
| rows.append(("Data Processing", "RAG Knowledge Store")) | |
| return rows | |
| def _generate_kpis(self) -> pd.DataFrame: | |
| now = self.generated_at.replace(second=0, microsecond=0) | |
| rows: List[Dict[str, Any]] = [] | |
| scenario_platform = self.scenario["platform"] | |
| scenario_domain = self.scenario["domain"] | |
| for minute_back in range(self.minutes, -1, -1): | |
| ts = now - timedelta(minutes=minute_back) | |
| anomaly_weight = max(0.0, 1.0 - (minute_back / 35.0)) if minute_back <= 35 else 0.0 | |
| for domain, platform in self._platform_rows(): | |
| baseline_latency = self.rng.uniform(25, 85) | |
| baseline_cpu = self.rng.uniform(22, 65) | |
| baseline_mem = self.rng.uniform(30, 72) | |
| baseline_error = self.rng.uniform(0.01, 0.40) | |
| baseline_loss = self.rng.uniform(0.00, 0.08) | |
| session_count = int(self.rng.uniform(5000, 45000)) | |
| is_focus = platform == scenario_platform | |
| is_neighbor = scenario_platform in ["AMF", "SMF", "PCF", "IMS", "E911 Router"] and platform in { | |
| "AMF": ["SMF", "AUSF", "Kafka Telemetry Pipeline"], | |
| "SMF": ["AMF", "UPF", "PCF"], | |
| "PCF": ["SMF", "Charging Gateway", "OCS"], | |
| "IMS": ["SBC", "HSS"], | |
| "E911 Router": ["PSAP Gateway", "ALI DB"], | |
| }.get(scenario_platform, []) | |
| impact = anomaly_weight * (1.0 if is_focus else 0.45 if is_neighbor else 0.0) | |
| latency = baseline_latency + impact * self.rng.uniform(80, 260) | |
| cpu = baseline_cpu + impact * self.rng.uniform(25, 50) | |
| memory = baseline_mem + impact * self.rng.uniform(10, 22) | |
| error_rate = baseline_error + impact * self.rng.uniform(1.2, 7.5) | |
| packet_loss = baseline_loss + impact * self.rng.uniform(0.15, 2.7) | |
| health_score = max( | |
| 1, | |
| min( | |
| 100, | |
| 100 | |
| - (latency / 8) | |
| - (cpu / 4) | |
| - (error_rate * 5) | |
| - (packet_loss * 4), | |
| ), | |
| ) | |
| rows.append( | |
| { | |
| "timestamp": ts, | |
| "domain": domain, | |
| "platform": platform, | |
| "latency_ms": round(latency, 2), | |
| "cpu_pct": round(min(cpu, 99.0), 2), | |
| "memory_pct": round(min(memory, 98.0), 2), | |
| "error_rate_pct": round(min(error_rate, 25.0), 3), | |
| "packet_loss_pct": round(min(packet_loss, 15.0), 3), | |
| "session_count": session_count + int(impact * self.rng.uniform(10000, 85000)), | |
| "health_score": round(health_score, 1), | |
| "scenario_focus": is_focus, | |
| } | |
| ) | |
| return pd.DataFrame(rows) | |
| def _generate_alerts(self) -> pd.DataFrame: | |
| now = self.generated_at.replace(second=0, microsecond=0) | |
| rows = [] | |
| scenario = self.scenario | |
| severities = ["Low", "Medium", "High", "Critical"] | |
| focus_severity = { | |
| "low": "Medium", | |
| "medium": "High", | |
| "high": "High", | |
| "critical": "Critical", | |
| }[scenario["risk_hint"]] | |
| rows.append( | |
| { | |
| "alert_id": f"AL-{self.seed}-{uuid.uuid4().hex[:6]}", | |
| "timestamp": now - timedelta(minutes=3), | |
| "domain": scenario["domain"], | |
| "platform": scenario["platform"], | |
| "severity": focus_severity, | |
| "title": self.scenario_name, | |
| "description": ", ".join(scenario["symptoms"]), | |
| "status": "Active", | |
| } | |
| ) | |
| for i in range(10): | |
| domain = self.rng.choice(DOMAINS) | |
| platform = self.rng.choice(PLATFORMS[domain]) | |
| severity = self.rng.choices(severities, weights=[35, 40, 20, 5], k=1)[0] | |
| rows.append( | |
| { | |
| "alert_id": f"AL-{self.seed}-{i:03d}", | |
| "timestamp": now - timedelta(minutes=self.rng.randint(4, 110)), | |
| "domain": domain, | |
| "platform": platform, | |
| "severity": severity, | |
| "title": f"{platform} {self.rng.choice(['Latency', 'Error Rate', 'Queue', 'Dependency'])} Alert", | |
| "description": self.rng.choice( | |
| [ | |
| "Threshold exceeded for two consecutive windows", | |
| "Intermittent dependency failures observed", | |
| "Customer-impacting KPI trending downward", | |
| "Telemetry anomaly detected by correlation model", | |
| ] | |
| ), | |
| "status": self.rng.choice(["Active", "Acknowledged", "Resolved"]), | |
| } | |
| ) | |
| return pd.DataFrame(rows).sort_values("timestamp", ascending=False).reset_index(drop=True) | |
| def _generate_tickets(self) -> pd.DataFrame: | |
| now = self.generated_at.replace(second=0, microsecond=0) | |
| rows = [] | |
| for i in range(16): | |
| domain = self.rng.choice(DOMAINS) | |
| platform = self.rng.choice(PLATFORMS[domain]) | |
| rows.append( | |
| { | |
| "ticket_id": f"INC{self.seed}{i:04d}", | |
| "opened_at": now - timedelta(hours=self.rng.randint(1, 72)), | |
| "domain": domain, | |
| "platform": platform, | |
| "priority": self.rng.choice(["P1", "P2", "P3", "P4"]), | |
| "status": self.rng.choice(["Open", "Investigating", "Resolved", "Monitoring"]), | |
| "summary": f"{platform} investigation: {self.rng.choice(['latency', 'errors', 'capacity', 'dependency'])}", | |
| } | |
| ) | |
| rows.append( | |
| { | |
| "ticket_id": f"INC{self.seed}FOCUS", | |
| "opened_at": now - timedelta(minutes=18), | |
| "domain": self.scenario["domain"], | |
| "platform": self.scenario["platform"], | |
| "priority": "P1" if self.scenario["risk_hint"] in ["high", "critical"] else "P2", | |
| "status": "Investigating", | |
| "summary": f"Current incident: {self.scenario_name}", | |
| } | |
| ) | |
| return pd.DataFrame(rows).sort_values("opened_at", ascending=False).reset_index(drop=True) | |
| def _generate_changes(self) -> pd.DataFrame: | |
| now = self.generated_at.replace(second=0, microsecond=0) | |
| rows = [] | |
| for i in range(14): | |
| domain = self.rng.choice(DOMAINS) | |
| platform = self.rng.choice(PLATFORMS[domain]) | |
| rows.append( | |
| { | |
| "change_id": f"CHG{self.seed}{i:04d}", | |
| "window_start": now - timedelta(hours=self.rng.randint(2, 96)), | |
| "domain": domain, | |
| "platform": platform, | |
| "change_type": self.rng.choice(["config", "software", "capacity", "routing", "policy"]), | |
| "risk": self.rng.choice(["Low", "Medium", "High"]), | |
| "status": self.rng.choice(["Completed", "Rolled Back", "Scheduled"]), | |
| "summary": f"{platform} {self.rng.choice(['maintenance', 'rule update', 'image update', 'routing update'])}", | |
| } | |
| ) | |
| if self.scenario["risk_hint"] in ["high", "critical"]: | |
| rows.append( | |
| { | |
| "change_id": f"CHG{self.seed}FOCUS", | |
| "window_start": now - timedelta(minutes=42), | |
| "domain": self.scenario["domain"], | |
| "platform": self.scenario["platform"], | |
| "change_type": "policy" if self.scenario["platform"] == "PCF" else "routing", | |
| "risk": "High", | |
| "status": "Completed", | |
| "summary": f"Recent high-risk change near {self.scenario['platform']}", | |
| } | |
| ) | |
| return pd.DataFrame(rows).sort_values("window_start", ascending=False).reset_index(drop=True) | |
| def _generate_topology(self) -> pd.DataFrame: | |
| edges = [ | |
| ("AMF", "SMF", "N11 signaling"), | |
| ("AMF", "AUSF", "subscriber authentication"), | |
| ("SMF", "UPF", "N4 session control"), | |
| ("SMF", "PCF", "policy lookup"), | |
| ("PCF", "OCS", "charging policy"), | |
| ("PCF", "Charging Gateway", "charging events"), | |
| ("IMS", "SBC", "SIP adjacency"), | |
| ("IMS", "HSS", "subscriber profile"), | |
| ("E911 Router", "PSAP Gateway", "emergency handoff"), | |
| ("E911 Router", "ALI DB", "location lookup"), | |
| ("SMSC", "IP-SM-GW", "SMS routing"), | |
| ("Kafka Telemetry Pipeline", "RAG Knowledge Store", "enriched telemetry"), | |
| ("Kafka Telemetry Pipeline", "Observability MCP Server", "metrics/traces/logs"), | |
| ("Runbook MCP Server", "RAG Knowledge Store", "SOP retrieval"), | |
| ("Ticketing MCP Server", "Change Mgmt MCP Server", "incident/change correlation"), | |
| ] | |
| return pd.DataFrame(edges, columns=["source", "target", "relationship"]) | |
| def _generate_runbooks(self) -> Dict[str, Dict[str, Any]]: | |
| return { | |
| "Capacity / control-plane overload": { | |
| "title": "Packet Core Overload Runbook", | |
| "steps": [ | |
| "Validate regional registration and attach KPIs.", | |
| "Check alarm burst and reconnect storm telemetry.", | |
| "Scale out low-risk control-plane pool if guardrail passes.", | |
| "Monitor recovery for two KPI windows.", | |
| ], | |
| "approved_actions": ["scale_out", "traffic_shift_10_percent", "monitor_only"], | |
| }, | |
| "Session management latency": { | |
| "title": "SMF Session Latency Runbook", | |
| "steps": [ | |
| "Check SMF PDU session setup latency.", | |
| "Correlate PCF and UPF dependency timings.", | |
| "Drain new sessions from degraded pool after approval.", | |
| "Validate session establishment and customer-impact metrics.", | |
| ], | |
| "approved_actions": ["drain_pool", "dependency_check", "monitor_only"], | |
| }, | |
| "Policy rule regression": { | |
| "title": "Policy and Charging Regression Runbook", | |
| "steps": [ | |
| "Identify recent PCF/PCRF rule changes.", | |
| "Compare failed decisions by subscriber cohort.", | |
| "Require human approval before rollback.", | |
| "Validate charging records and policy response codes.", | |
| ], | |
| "approved_actions": ["human_approval_required", "rollback_policy_bundle", "monitor_only"], | |
| }, | |
| "Emergency routing risk": { | |
| "title": "E911 Routing Safety Runbook", | |
| "steps": [ | |
| "Declare critical operational safety incident.", | |
| "Escalate to E911 operations immediately.", | |
| "Freeze autonomous execution.", | |
| "Manually validate PSAP routes and location database consistency.", | |
| ], | |
| "approved_actions": ["escalate_only", "manual_validation", "no_auto_execute"], | |
| }, | |
| "Voice call setup degradation": { | |
| "title": "IMS Voice Degradation Runbook", | |
| "steps": [ | |
| "Validate SIP response code distribution.", | |
| "Check SBC adjacency health.", | |
| "Route limited traffic through healthy SBC pair after approval.", | |
| "Monitor call completion and drop rates.", | |
| ], | |
| "approved_actions": ["limited_traffic_shift", "sbc_health_check", "monitor_only"], | |
| }, | |
| "Observability pipeline backlog": { | |
| "title": "Telemetry Pipeline Backlog Runbook", | |
| "steps": [ | |
| "Check Kafka consumer lag by topic.", | |
| "Scale consumers if lag affects alert freshness.", | |
| "Throttle non-critical enrichment jobs.", | |
| "Confirm near-real-time alerting restored.", | |
| ], | |
| "approved_actions": ["scale_consumers", "throttle_enrichment", "monitor_only"], | |
| }, | |
| } | |
| def append_audit(self, actor: str, event: str, status: str, detail: str) -> None: | |
| row = pd.DataFrame( | |
| [ | |
| { | |
| "timestamp": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"), | |
| "actor": actor, | |
| "event": event, | |
| "status": status, | |
| "detail": detail, | |
| } | |
| ] | |
| ) | |
| self.audit_df = pd.concat([row, self.audit_df], ignore_index=True) | |
| # ----------------------------------------------------------------------------- | |
| # MCP-style tool framework | |
| # ----------------------------------------------------------------------------- | |
| class MCPTool: | |
| name: str | |
| server: str | |
| description: str | |
| input_schema: Dict[str, str] | |
| output_schema: Dict[str, str] | |
| permission_tier: str | |
| risk_tier: str | |
| func: Callable[..., Dict[str, Any]] | |
| class ToolCallRecord: | |
| timestamp: str | |
| tool: str | |
| server: str | |
| input: str | |
| output_summary: str | |
| permission_tier: str | |
| risk_tier: str | |
| status: str | |
| class MCPToolRegistry: | |
| """Simple MCP-shaped registry that records every tool call.""" | |
| def __init__(self, dataset: SyntheticTelecomDataset): | |
| self.dataset = dataset | |
| self.tools: Dict[str, MCPTool] = {} | |
| self.call_log: List[ToolCallRecord] = [] | |
| self._register_default_tools() | |
| def register(self, tool: MCPTool) -> None: | |
| self.tools[tool.name] = tool | |
| def call(self, name: str, **kwargs: Any) -> Dict[str, Any]: | |
| if name not in self.tools: | |
| raise KeyError(f"Tool not registered: {name}") | |
| tool = self.tools[name] | |
| try: | |
| result = tool.func(**kwargs) | |
| status = "Success" | |
| except Exception as exc: # pragma: no cover - demo safety | |
| result = {"error": str(exc)} | |
| status = "Error" | |
| output_summary = self._summarize_output(result) | |
| self.call_log.append( | |
| ToolCallRecord( | |
| timestamp=datetime.utcnow().strftime("%H:%M:%S"), | |
| tool=tool.name, | |
| server=tool.server, | |
| input=json.dumps(kwargs, default=str), | |
| output_summary=output_summary, | |
| permission_tier=tool.permission_tier, | |
| risk_tier=tool.risk_tier, | |
| status=status, | |
| ) | |
| ) | |
| self.dataset.append_audit( | |
| actor="MCP Server Gateway", | |
| event=f"Tool invoked: {tool.name}", | |
| status=status, | |
| detail=output_summary, | |
| ) | |
| return result | |
| def _summarize_output(result: Dict[str, Any]) -> str: | |
| if "summary" in result: | |
| return str(result["summary"]) | |
| if "error" in result: | |
| return str(result["error"]) | |
| return textwrap.shorten(json.dumps(result, default=str), width=120, placeholder="...") | |
| def calls_df(self) -> pd.DataFrame: | |
| return pd.DataFrame([record.__dict__ for record in self.call_log]) | |
| def registry_df(self) -> pd.DataFrame: | |
| rows = [] | |
| for tool in self.tools.values(): | |
| rows.append( | |
| { | |
| "tool_name": tool.name, | |
| "mcp_server": tool.server, | |
| "description": tool.description, | |
| "permission_tier": tool.permission_tier, | |
| "risk_tier": tool.risk_tier, | |
| "input_schema": json.dumps(tool.input_schema), | |
| "output_schema": json.dumps(tool.output_schema), | |
| } | |
| ) | |
| return pd.DataFrame(rows) | |
| def _register_default_tools(self) -> None: | |
| ds = self.dataset | |
| def get_kpi_snapshot(platform: str, minutes: int = 30) -> Dict[str, Any]: | |
| cutoff = ds.generated_at - timedelta(minutes=minutes) | |
| df = ds.kpi_df[(ds.kpi_df["platform"] == platform) & (ds.kpi_df["timestamp"] >= cutoff)] | |
| if df.empty: | |
| return {"summary": f"No KPI data found for {platform}", "records": []} | |
| latest = df.sort_values("timestamp").tail(1).iloc[0].to_dict() | |
| trend = { | |
| "latency_ms_avg": round(df["latency_ms"].mean(), 2), | |
| "error_rate_pct_avg": round(df["error_rate_pct"].mean(), 3), | |
| "cpu_pct_avg": round(df["cpu_pct"].mean(), 2), | |
| "health_score_latest": latest["health_score"], | |
| "health_score_min": round(df["health_score"].min(), 1), | |
| } | |
| return { | |
| "summary": f"{platform} health {latest['health_score']} with avg latency {trend['latency_ms_avg']} ms", | |
| "latest": latest, | |
| "trend": trend, | |
| } | |
| def stream_alerts(platform: str, limit: int = 8) -> Dict[str, Any]: | |
| df = ds.alerts_df[(ds.alerts_df["platform"] == platform) | (ds.alerts_df["description"].str.contains(platform, case=False, na=False))] | |
| if df.empty: | |
| df = ds.alerts_df.head(limit) | |
| return { | |
| "summary": f"Returned {min(limit, len(df))} active/recent alerts", | |
| "alerts": df.head(limit).to_dict(orient="records"), | |
| } | |
| def retrieve_runbook(fault_type: str) -> Dict[str, Any]: | |
| runbook = ds.runbooks.get(fault_type, ds.runbooks["Capacity / control-plane overload"]) | |
| return { | |
| "summary": f"Runbook: {runbook['title']}", | |
| "runbook": runbook, | |
| } | |
| def get_topology_neighbors(platform: str) -> Dict[str, Any]: | |
| df = ds.topology_df[(ds.topology_df["source"] == platform) | (ds.topology_df["target"] == platform)] | |
| neighbors = sorted(set(df["source"].tolist() + df["target"].tolist()) - {platform}) | |
| return { | |
| "summary": f"{platform} neighbors: {', '.join(neighbors) if neighbors else 'none'}", | |
| "neighbors": neighbors, | |
| "edges": df.to_dict(orient="records"), | |
| } | |
| def search_related_tickets(platform: str, hours: int = 72) -> Dict[str, Any]: | |
| cutoff = ds.generated_at - timedelta(hours=hours) | |
| df = ds.tickets_df[(ds.tickets_df["platform"] == platform) & (ds.tickets_df["opened_at"] >= cutoff)] | |
| return { | |
| "summary": f"Found {len(df)} related tickets in {hours} hours", | |
| "tickets": df.to_dict(orient="records"), | |
| } | |
| def check_recent_changes(platform: str, hours: int = 72) -> Dict[str, Any]: | |
| cutoff = ds.generated_at - timedelta(hours=hours) | |
| df = ds.changes_df[(ds.changes_df["platform"] == platform) & (ds.changes_df["window_start"] >= cutoff)] | |
| return { | |
| "summary": f"Found {len(df)} recent changes for {platform}", | |
| "changes": df.to_dict(orient="records"), | |
| } | |
| def correlate_signals(platform: str, fault_type: str) -> Dict[str, Any]: | |
| kpis = get_kpi_snapshot(platform, minutes=35) | |
| alerts = stream_alerts(platform, limit=5) | |
| changes = check_recent_changes(platform, hours=6) | |
| tickets = search_related_tickets(platform, hours=72) | |
| evidence = [] | |
| trend = kpis.get("trend", {}) | |
| if trend.get("health_score_min", 100) < 60: | |
| evidence.append("Health score dropped below 60 in the last 35 minutes") | |
| if trend.get("latency_ms_avg", 0) > 120: | |
| evidence.append("Latency average is above the operating baseline") | |
| if len(alerts.get("alerts", [])) > 0: | |
| evidence.append("Active/recent alerts exist for the impacted platform") | |
| if len(changes.get("changes", [])) > 0: | |
| evidence.append("Recent change activity exists near the incident window") | |
| if len(tickets.get("tickets", [])) > 0: | |
| evidence.append("Related incident tickets exist") | |
| confidence = min(97, 55 + 8 * len(evidence)) | |
| return { | |
| "summary": f"Correlation confidence {confidence}% with {len(evidence)} evidence items", | |
| "confidence": confidence, | |
| "evidence": evidence, | |
| "fault_type": fault_type, | |
| } | |
| def guardrail_risk_score(action: str, platform: str, fault_type: str) -> Dict[str, Any]: | |
| scenario_risk = ds.scenario["risk_hint"] | |
| base = {"low": 22, "medium": 48, "high": 74, "critical": 96}[scenario_risk] | |
| action_lower = action.lower() | |
| if any(term in action_lower for term in ["rollback", "routing", "e911", "emergency", "route ", "traffic"]): | |
| base += 14 | |
| if any(term in action_lower for term in ["scale", "consumer"]): | |
| base -= 18 | |
| if action_lower.strip().startswith("monitor only"): | |
| base -= 12 | |
| if platform in ["E911 Router", "PSAP Gateway", "ALI DB"]: | |
| base = max(base, 90) | |
| score = max(1, min(99, base)) | |
| if score < 35: | |
| decision = "Auto-executable with guardrails" | |
| risk_level = "Low" | |
| elif score < 70: | |
| decision = "Human approval required" | |
| risk_level = "Medium" | |
| elif score < 90: | |
| decision = "Senior operator approval required" | |
| risk_level = "High" | |
| else: | |
| decision = "Escalate only; autonomous execution blocked" | |
| risk_level = "Critical" | |
| return { | |
| "summary": f"Risk {risk_level} ({score}/100): {decision}", | |
| "risk_score": score, | |
| "risk_level": risk_level, | |
| "decision": decision, | |
| "blocked": score >= 90, | |
| } | |
| def propose_mitigation(platform: str, fault_type: str) -> Dict[str, Any]: | |
| action = ds.scenario["recommended_action"] | |
| return { | |
| "summary": f"Recommended action generated for {platform}", | |
| "recommended_action": action, | |
| "alternatives": [ | |
| "Monitor only for one additional KPI window", | |
| "Create/escalate ticket with incident summary", | |
| "Run dependency health validation before mitigation", | |
| ], | |
| } | |
| def create_incident_summary(platform: str, scenario_name: str, confidence: int, risk_level: str) -> Dict[str, Any]: | |
| summary = ( | |
| f"{scenario_name} affecting {platform}. The supervisor correlated KPI degradation, " | |
| f"alerts, related tickets, topology, and change data. Confidence: {confidence}%. " | |
| f"Guardrail risk: {risk_level}." | |
| ) | |
| return {"summary": "Incident summary created", "incident_summary": summary} | |
| def create_ticket(platform: str, incident_summary: str, priority: str) -> Dict[str, Any]: | |
| ticket_id = f"INC-DEMO-{uuid.uuid4().hex[:8].upper()}" | |
| return { | |
| "summary": f"Created simulated ticket {ticket_id}", | |
| "ticket_id": ticket_id, | |
| "platform": platform, | |
| "priority": priority, | |
| "status": "Created - simulated", | |
| "incident_summary": incident_summary, | |
| } | |
| def execute_low_risk_action(platform: str, action: str, approval: str) -> Dict[str, Any]: | |
| if approval != "approved_low_risk": | |
| return { | |
| "summary": "Execution blocked because approval token is missing or risk is not low", | |
| "executed": False, | |
| } | |
| return { | |
| "summary": f"Simulated execution completed for {platform}", | |
| "executed": True, | |
| "action": action, | |
| "validation": "Synthetic health check improved in simulated follow-up window", | |
| } | |
| self.register( | |
| MCPTool( | |
| name="get_kpi_snapshot", | |
| server="Observability MCP Server", | |
| description="Retrieves recent KPI metrics for a wireless core platform.", | |
| input_schema={"platform": "str", "minutes": "int"}, | |
| output_schema={"latest": "dict", "trend": "dict"}, | |
| permission_tier="read", | |
| risk_tier="safe", | |
| func=get_kpi_snapshot, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="stream_alerts", | |
| server="Kafka MCP Server", | |
| description="Streams active and recent alarms/events for the affected platform.", | |
| input_schema={"platform": "str", "limit": "int"}, | |
| output_schema={"alerts": "list"}, | |
| permission_tier="read", | |
| risk_tier="safe", | |
| func=stream_alerts, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="retrieve_runbook", | |
| server="Runbook MCP Server", | |
| description="Retrieves SOP/runbook content from a knowledge store.", | |
| input_schema={"fault_type": "str"}, | |
| output_schema={"runbook": "dict"}, | |
| permission_tier="read", | |
| risk_tier="safe", | |
| func=retrieve_runbook, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="get_topology_neighbors", | |
| server="Topology MCP Server", | |
| description="Retrieves network dependency edges around a platform.", | |
| input_schema={"platform": "str"}, | |
| output_schema={"neighbors": "list", "edges": "list"}, | |
| permission_tier="read", | |
| risk_tier="safe", | |
| func=get_topology_neighbors, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="search_related_tickets", | |
| server="Ticketing MCP Server", | |
| description="Searches related incidents and trouble tickets.", | |
| input_schema={"platform": "str", "hours": "int"}, | |
| output_schema={"tickets": "list"}, | |
| permission_tier="read", | |
| risk_tier="safe", | |
| func=search_related_tickets, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="check_recent_changes", | |
| server="Change Mgmt MCP Server", | |
| description="Finds recent change records that may explain a fault.", | |
| input_schema={"platform": "str", "hours": "int"}, | |
| output_schema={"changes": "list"}, | |
| permission_tier="read", | |
| risk_tier="safe", | |
| func=check_recent_changes, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="correlate_signals", | |
| server="Correlation MCP Server", | |
| description="Combines KPIs, alerts, tickets, changes, and topology into evidence.", | |
| input_schema={"platform": "str", "fault_type": "str"}, | |
| output_schema={"confidence": "int", "evidence": "list"}, | |
| permission_tier="read", | |
| risk_tier="safe", | |
| func=correlate_signals, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="guardrail_risk_score", | |
| server="Guardrail Gateway", | |
| description="Scores operational risk before mitigation or execution.", | |
| input_schema={"action": "str", "platform": "str", "fault_type": "str"}, | |
| output_schema={"risk_score": "int", "decision": "str", "blocked": "bool"}, | |
| permission_tier="approval_gate", | |
| risk_tier="control", | |
| func=guardrail_risk_score, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="propose_mitigation", | |
| server="Mitigation MCP Server", | |
| description="Generates mitigation recommendations from runbook and evidence.", | |
| input_schema={"platform": "str", "fault_type": "str"}, | |
| output_schema={"recommended_action": "str", "alternatives": "list"}, | |
| permission_tier="read/recommend", | |
| risk_tier="advisory", | |
| func=propose_mitigation, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="create_incident_summary", | |
| server="Reporting MCP Server", | |
| description="Creates incident summary and RCA narrative.", | |
| input_schema={"platform": "str", "scenario_name": "str", "confidence": "int", "risk_level": "str"}, | |
| output_schema={"incident_summary": "str"}, | |
| permission_tier="write_report", | |
| risk_tier="safe", | |
| func=create_incident_summary, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="create_ticket", | |
| server="Ticketing MCP Server", | |
| description="Creates a simulated escalation ticket.", | |
| input_schema={"platform": "str", "incident_summary": "str", "priority": "str"}, | |
| output_schema={"ticket_id": "str", "status": "str"}, | |
| permission_tier="write_ticket", | |
| risk_tier="controlled", | |
| func=create_ticket, | |
| ) | |
| ) | |
| self.register( | |
| MCPTool( | |
| name="execute_low_risk_action", | |
| server="Network Action MCP Server", | |
| description="Simulates approved low-risk action execution.", | |
| input_schema={"platform": "str", "action": "str", "approval": "str"}, | |
| output_schema={"executed": "bool", "validation": "str"}, | |
| permission_tier="execute_guardrailed", | |
| risk_tier="highly_controlled", | |
| func=execute_low_risk_action, | |
| ) | |
| ) | |
| # ----------------------------------------------------------------------------- | |
| # Supervisor agent model | |
| # ----------------------------------------------------------------------------- | |
| class SupervisorResult: | |
| scenario_name: str | |
| platform: str | |
| domain: str | |
| fault_type: str | |
| evidence: List[str] | |
| confidence: int | |
| root_cause: str | |
| recommended_action: str | |
| risk_score: int | |
| risk_level: str | |
| guardrail_decision: str | |
| execution_status: str | |
| incident_summary: str | |
| ticket_id: str | |
| runbook_title: str | |
| runbook_steps: List[str] | |
| specialized_agents: List[Dict[str, str]] | |
| class SupervisorAgent: | |
| """ | |
| Supervisor agent that plans, routes, monitors, and gates specialized agents. | |
| This intentionally acts as a deterministic supervisor for demos. In an enterprise | |
| deployment, each specialist could call Claude/Copilot/LLM tools, while the supervisor | |
| enforces routing, evidence capture, guardrails, and approval state. | |
| """ | |
| def __init__(self, registry: MCPToolRegistry, dataset: SyntheticTelecomDataset): | |
| self.registry = registry | |
| self.dataset = dataset | |
| def run(self, execute_when_allowed: bool = False) -> SupervisorResult: | |
| s = self.dataset.scenario | |
| platform = s["platform"] | |
| fault_type = s["fault_type"] | |
| self.dataset.append_audit("Supervisor Agent", "Plan", "Started", "Created incident analysis plan") | |
| # Specialized agents routed by supervisor. | |
| kpi = self.registry.call("get_kpi_snapshot", platform=platform, minutes=35) | |
| alerts = self.registry.call("stream_alerts", platform=platform, limit=6) | |
| runbook = self.registry.call("retrieve_runbook", fault_type=fault_type) | |
| topology = self.registry.call("get_topology_neighbors", platform=platform) | |
| tickets = self.registry.call("search_related_tickets", platform=platform, hours=72) | |
| changes = self.registry.call("check_recent_changes", platform=platform, hours=8) | |
| correlation = self.registry.call("correlate_signals", platform=platform, fault_type=fault_type) | |
| mitigation = self.registry.call("propose_mitigation", platform=platform, fault_type=fault_type) | |
| risk = self.registry.call( | |
| "guardrail_risk_score", | |
| action=mitigation["recommended_action"], | |
| platform=platform, | |
| fault_type=fault_type, | |
| ) | |
| summary = self.registry.call( | |
| "create_incident_summary", | |
| platform=platform, | |
| scenario_name=self.dataset.scenario_name, | |
| confidence=correlation["confidence"], | |
| risk_level=risk["risk_level"], | |
| ) | |
| priority = "P1" if risk["risk_level"] in ["High", "Critical"] else "P2" | |
| ticket = self.registry.call( | |
| "create_ticket", | |
| platform=platform, | |
| incident_summary=summary["incident_summary"], | |
| priority=priority, | |
| ) | |
| execution_status = "Recommendation only; no action executed." | |
| if execute_when_allowed and risk["risk_level"] == "Low" and not risk["blocked"]: | |
| execution = self.registry.call( | |
| "execute_low_risk_action", | |
| platform=platform, | |
| action=mitigation["recommended_action"], | |
| approval="approved_low_risk", | |
| ) | |
| execution_status = execution.get("summary", execution_status) | |
| elif execute_when_allowed: | |
| execution_status = f"Execution blocked by guardrail: {risk['decision']}" | |
| self.dataset.append_audit("Guardrail Gateway", "Execution Blocked", "Blocked", execution_status) | |
| specialized_agents = [ | |
| {"agent": "KPI Monitoring Agent", "responsibility": "Read health, latency, CPU, memory, errors", "status": kpi["summary"]}, | |
| {"agent": "Fault Detection Agent", "responsibility": "Interpret alarms and event bursts", "status": alerts["summary"]}, | |
| {"agent": "RCA Agent", "responsibility": "Correlate signals into evidence", "status": correlation["summary"]}, | |
| {"agent": "Fault Isolation Agent", "responsibility": "Use topology to find neighbors", "status": topology["summary"]}, | |
| {"agent": "Change Risk Agent", "responsibility": "Check change windows and regressions", "status": changes["summary"]}, | |
| {"agent": "Mitigation Advisor Agent", "responsibility": "Recommend runbook-safe action", "status": mitigation["summary"]}, | |
| {"agent": "Guardrail Agent", "responsibility": "Block unsafe actions", "status": risk["summary"]}, | |
| {"agent": "Reporting Agent", "responsibility": "Create ticket and executive summary", "status": ticket["summary"]}, | |
| ] | |
| self.dataset.append_audit("Supervisor Agent", "Plan", "Completed", "Specialized agents completed with guardrail review") | |
| return SupervisorResult( | |
| scenario_name=self.dataset.scenario_name, | |
| platform=platform, | |
| domain=s["domain"], | |
| fault_type=fault_type, | |
| evidence=correlation["evidence"], | |
| confidence=int(correlation["confidence"]), | |
| root_cause=s["expected_root_cause"], | |
| recommended_action=mitigation["recommended_action"], | |
| risk_score=int(risk["risk_score"]), | |
| risk_level=risk["risk_level"], | |
| guardrail_decision=risk["decision"], | |
| execution_status=execution_status, | |
| incident_summary=summary["incident_summary"], | |
| ticket_id=ticket["ticket_id"], | |
| runbook_title=runbook["runbook"]["title"], | |
| runbook_steps=runbook["runbook"]["steps"], | |
| specialized_agents=specialized_agents, | |
| ) | |
| # ----------------------------------------------------------------------------- | |
| # Visualization and Gradio callbacks | |
| # ----------------------------------------------------------------------------- | |
| def status_badge(label: str, value: str, color: str = "#eef2ff") -> str: | |
| return f""" | |
| <div class='status-card'> | |
| <div class='status-label'>{label}</div> | |
| <div class='status-value'>{value}</div> | |
| </div> | |
| """ | |
| def build_cards(result: SupervisorResult) -> str: | |
| risk_class = { | |
| "Low": "risk-low", | |
| "Medium": "risk-med", | |
| "High": "risk-high", | |
| "Critical": "risk-critical", | |
| }.get(result.risk_level, "risk-med") | |
| return f""" | |
| <div class="cards"> | |
| <div class="card"><div class="k">Scenario</div><div class="v">{result.scenario_name}</div></div> | |
| <div class="card"><div class="k">Platform</div><div class="v">{result.platform}</div><div class="s">{result.domain}</div></div> | |
| <div class="card"><div class="k">Confidence</div><div class="v">{result.confidence}%</div><div class="s">Evidence-based RCA</div></div> | |
| <div class="card {risk_class}"><div class="k">Guardrail Risk</div><div class="v">{result.risk_level}</div><div class="s">Score {result.risk_score}/100</div></div> | |
| <div class="card"><div class="k">Ticket</div><div class="v">{result.ticket_id}</div><div class="s">Simulated ticketing MCP</div></div> | |
| </div> | |
| """ | |
| def make_kpi_plot(dataset: SyntheticTelecomDataset, platform: str) -> go.Figure: | |
| df = dataset.kpi_df[dataset.kpi_df["platform"] == platform].sort_values("timestamp") | |
| if df.empty: | |
| fig = go.Figure() | |
| fig.update_layout(title="No KPI data found") | |
| return fig | |
| fig = go.Figure() | |
| fig.add_trace( | |
| go.Scatter( | |
| x=df["timestamp"], | |
| y=df["health_score"], | |
| mode="lines+markers", | |
| name="Health Score", | |
| yaxis="y1", | |
| ) | |
| ) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=df["timestamp"], | |
| y=df["latency_ms"], | |
| mode="lines", | |
| name="Latency ms", | |
| yaxis="y2", | |
| ) | |
| ) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=df["timestamp"], | |
| y=df["error_rate_pct"], | |
| mode="lines", | |
| name="Error rate %", | |
| yaxis="y3", | |
| ) | |
| ) | |
| fig.update_layout( | |
| title=f"Synthetic KPI timeline: {platform}", | |
| legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0), | |
| margin=dict(l=30, r=30, t=70, b=30), | |
| yaxis=dict(title="Health", range=[0, 105]), | |
| yaxis2=dict(title="Latency", overlaying="y", side="right"), | |
| yaxis3=dict(title="Error %", overlaying="y", side="right", anchor="free", position=0.95, showgrid=False), | |
| height=420, | |
| ) | |
| return fig | |
| def make_topology_plot(dataset: SyntheticTelecomDataset, platform: str) -> go.Figure: | |
| df = dataset.topology_df[(dataset.topology_df["source"] == platform) | (dataset.topology_df["target"] == platform)] | |
| nodes = sorted(set(df["source"].tolist() + df["target"].tolist())) | |
| if not nodes: | |
| nodes = [platform] | |
| # Lightweight radial layout without networkx dependency. | |
| positions: Dict[str, Tuple[float, float]] = {platform: (0, 0)} | |
| others = [n for n in nodes if n != platform] | |
| for i, node in enumerate(others): | |
| angle = 2 * math.pi * i / max(len(others), 1) | |
| positions[node] = (math.cos(angle), math.sin(angle)) | |
| edge_x, edge_y = [], [] | |
| edge_text = [] | |
| for _, row in df.iterrows(): | |
| x0, y0 = positions[row["source"]] | |
| x1, y1 = positions[row["target"]] | |
| edge_x += [x0, x1, None] | |
| edge_y += [y0, y1, None] | |
| edge_text.append(row["relationship"]) | |
| fig = go.Figure() | |
| fig.add_trace(go.Scatter(x=edge_x, y=edge_y, mode="lines", hoverinfo="none", name="Dependency")) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=[positions[n][0] for n in nodes], | |
| y=[positions[n][1] for n in nodes], | |
| mode="markers+text", | |
| text=nodes, | |
| textposition="bottom center", | |
| marker=dict(size=[28 if n == platform else 18 for n in nodes]), | |
| name="Platforms", | |
| ) | |
| ) | |
| fig.update_layout( | |
| title=f"Topology neighbors for {platform}", | |
| showlegend=False, | |
| height=360, | |
| margin=dict(l=20, r=20, t=60, b=20), | |
| xaxis=dict(showgrid=False, zeroline=False, visible=False), | |
| yaxis=dict(showgrid=False, zeroline=False, visible=False), | |
| ) | |
| return fig | |
| def result_markdown(result: SupervisorResult) -> str: | |
| evidence_md = "\n".join([f"- {item}" for item in result.evidence]) or "- No evidence captured" | |
| steps_md = "\n".join([f"{i+1}. {step}" for i, step in enumerate(result.runbook_steps)]) | |
| return f""" | |
| ### Supervisor Decision | |
| **Root cause hypothesis:** {result.root_cause} | |
| **Recommended action:** {result.recommended_action} | |
| **Guardrail decision:** {result.guardrail_decision} | |
| **Execution status:** {result.execution_status} | |
| ### Evidence collected by specialist agents | |
| {evidence_md} | |
| ### Runbook used: {result.runbook_title} | |
| {steps_md} | |
| ### Incident summary | |
| {result.incident_summary} | |
| """ | |
| def normalize_df_for_display(df: pd.DataFrame, limit: int = 50) -> pd.DataFrame: | |
| display = df.head(limit).copy() | |
| for col in display.columns: | |
| if pd.api.types.is_datetime64_any_dtype(display[col]): | |
| display[col] = display[col].dt.strftime("%Y-%m-%d %H:%M:%S") | |
| return display | |
| def run_supervisor_demo( | |
| scenario_name: str, | |
| seed: int, | |
| minutes: int, | |
| execute_when_allowed: bool, | |
| ) -> Tuple[str, go.Figure, go.Figure, str, pd.DataFrame, pd.DataFrame, pd.DataFrame, str]: | |
| dataset = SyntheticTelecomDataset(seed=int(seed), minutes=int(minutes), scenario_name=scenario_name) | |
| registry = MCPToolRegistry(dataset) | |
| supervisor = SupervisorAgent(registry, dataset) | |
| result = supervisor.run(execute_when_allowed=execute_when_allowed) | |
| cards = build_cards(result) | |
| kpi_plot = make_kpi_plot(dataset, result.platform) | |
| topology_plot = make_topology_plot(dataset, result.platform) | |
| md = result_markdown(result) | |
| agents_df = pd.DataFrame(result.specialized_agents) | |
| calls_df = registry.calls_df() | |
| audit_df = dataset.audit_df | |
| rca_json = json.dumps(result.__dict__, default=str, indent=2) | |
| return cards, kpi_plot, topology_plot, md, agents_df, calls_df, audit_df, rca_json | |
| def generate_data_preview(scenario_name: str, seed: int, minutes: int, platform_filter: str) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]: | |
| dataset = SyntheticTelecomDataset(seed=int(seed), minutes=int(minutes), scenario_name=scenario_name) | |
| kpi_df = dataset.kpi_df.copy() | |
| if platform_filter and platform_filter != "All": | |
| kpi_df = kpi_df[kpi_df["platform"] == platform_filter] | |
| return ( | |
| normalize_df_for_display(kpi_df.sort_values("timestamp", ascending=False), limit=100), | |
| normalize_df_for_display(dataset.alerts_df, limit=100), | |
| normalize_df_for_display(dataset.tickets_df, limit=100), | |
| normalize_df_for_display(dataset.changes_df, limit=100), | |
| ) | |
| def get_registry_preview(scenario_name: str, seed: int) -> pd.DataFrame: | |
| dataset = SyntheticTelecomDataset(seed=int(seed), minutes=120, scenario_name=scenario_name) | |
| registry = MCPToolRegistry(dataset) | |
| return registry.registry_df() | |
| def run_batch_tests(seed: int, execute_when_allowed: bool) -> Tuple[pd.DataFrame, str]: | |
| rows = [] | |
| for i, scenario_name in enumerate(SCENARIOS.keys()): | |
| dataset = SyntheticTelecomDataset(seed=int(seed) + i, minutes=120, scenario_name=scenario_name) | |
| registry = MCPToolRegistry(dataset) | |
| result = SupervisorAgent(registry, dataset).run(execute_when_allowed=execute_when_allowed) | |
| rows.append( | |
| { | |
| "scenario": scenario_name, | |
| "platform": result.platform, | |
| "fault_type": result.fault_type, | |
| "confidence": result.confidence, | |
| "risk_level": result.risk_level, | |
| "risk_score": result.risk_score, | |
| "guardrail_decision": result.guardrail_decision, | |
| "execution_status": result.execution_status, | |
| "tool_calls": len(registry.call_log), | |
| "ticket_id": result.ticket_id, | |
| } | |
| ) | |
| df = pd.DataFrame(rows) | |
| summary = ( | |
| f"Batch test completed across {len(df)} scenarios. " | |
| f"Low-risk auto-executable scenarios: {(df['risk_level'] == 'Low').sum()}. " | |
| f"Human/senior approval scenarios: {df['risk_level'].isin(['Medium', 'High']).sum()}. " | |
| f"Blocked critical scenarios: {(df['risk_level'] == 'Critical').sum()}." | |
| ) | |
| return df, summary | |
| def supervisor_architecture_md() -> str: | |
| return """ | |
| ## Supervisor model design | |
| The supervisor agent does **not** directly execute every task. It plans the incident workflow, calls specialist agents through MCP-style tools, captures evidence, applies risk gates, and then decides whether the result is recommendation-only, human-approved, or low-risk executable. | |
| ### Flow used in this demo | |
| 1. **Plan** — identify impacted wireless core platform and suspected fault type. | |
| 2. **Collect evidence** — call MCP tools for KPIs, alerts, topology, tickets, changes, and runbooks. | |
| 3. **Route specialist agents** — KPI Monitoring, Fault Detection, Fault Isolation, RCA, Mitigation Advisor, Change Risk, Guardrail, Human Approval, and Reporting. | |
| 4. **Guardrail before action** — score operational risk before any execution path. | |
| 5. **Respond** — create incident summary, RCA, ticket, and recommended action. | |
| 6. **Audit loop** — every MCP call is logged for traceability. | |
| ### MCP tooling pattern | |
| Each enterprise system is wrapped as an MCP server/tool boundary: | |
| - **Observability MCP Server:** metrics, logs, traces, dashboards | |
| - **Kafka MCP Server:** alarms, events, streaming telemetry | |
| - **Ticketing MCP Server:** incidents, tickets, problems | |
| - **Runbook MCP Server:** SOPs, mitigations, operational knowledge | |
| - **Topology MCP Server:** network dependencies and inventory | |
| - **Change Mgmt MCP Server:** recent changes and maintenance windows | |
| - **Guardrail Gateway:** risk scoring, approval state, execution policy | |
| - **Network Action MCP Server:** approved low-risk execution only | |
| This creates a clean separation between AI reasoning and operational systems. | |
| """ | |
| CUSTOM_CSS = """ | |
| #title h1 {font-size: 2.4rem; margin-bottom: 0.2rem;} | |
| #title p {font-size: 1.05rem; color: #4b5563;} | |
| .cards {display: grid; grid-template-columns: repeat(5, minmax(160px, 1fr)); gap: 12px; margin: 12px 0;} | |
| .card {background: white; border: 1px solid #e5e7eb; border-radius: 16px; padding: 14px; box-shadow: 0 2px 8px rgba(0,0,0,0.04);} | |
| .card .k {font-size: 0.78rem; color: #6b7280; text-transform: uppercase; letter-spacing: .03em;} | |
| .card .v {font-size: 1.15rem; font-weight: 700; color: #111827; margin-top: 4px;} | |
| .card .s {font-size: 0.82rem; color: #4b5563; margin-top: 4px;} | |
| .risk-low {border-left: 6px solid #16a34a;} | |
| .risk-med {border-left: 6px solid #d97706;} | |
| .risk-high {border-left: 6px solid #dc2626;} | |
| .risk-critical {border-left: 6px solid #7f1d1d; background: #fff7f7;} | |
| @media (max-width: 900px) {.cards {grid-template-columns: repeat(1, minmax(160px, 1fr));}} | |
| """ | |
| def build_app() -> gr.Blocks: | |
| all_platforms = ["All"] + sorted({p for platforms in PLATFORMS.values() for p in platforms} | {"Kafka Telemetry Pipeline", "RAG Knowledge Store"}) | |
| with gr.Blocks(css=CUSTOM_CSS, theme=gr.themes.Soft()) as demo: | |
| gr.HTML( | |
| f""" | |
| <div id='title'> | |
| <h1>{APP_TITLE}</h1> | |
| <p>{APP_SUBTITLE}</p> | |
| </div> | |
| """ | |
| ) | |
| gr.Markdown( | |
| "This is a self-contained Hugging Face demo. It uses synthetic wireless-core operations data and a deterministic supervisor agent so you can test the architecture without enterprise credentials." | |
| ) | |
| with gr.Row(): | |
| scenario = gr.Dropdown( | |
| choices=list(SCENARIOS.keys()), | |
| value="AMF Overload Detected", | |
| label="Synthetic incident scenario", | |
| ) | |
| seed = gr.Number(value=42, label="Synthetic data seed", precision=0) | |
| minutes = gr.Slider(30, 240, value=120, step=15, label="KPI history minutes") | |
| execute = gr.Checkbox(value=False, label="Simulate execution when guardrail allows low risk") | |
| with gr.Tabs(): | |
| with gr.Tab("Supervisor Operations Console"): | |
| run_btn = gr.Button("Run Supervisor Agent Flow", variant="primary") | |
| cards = gr.HTML() | |
| with gr.Row(): | |
| kpi_plot = gr.Plot(label="KPI Timeline") | |
| topology_plot = gr.Plot(label="Topology Neighbors") | |
| decision_md = gr.Markdown() | |
| agents_df = gr.Dataframe(label="Specialized agents routed by supervisor", wrap=True) | |
| with gr.Tab("MCP Tool Trace"): | |
| gr.Markdown("Every supervisor step calls an MCP-style tool. This table is the audit trace for tool routing, input, output summary, permission tier, and risk tier.") | |
| calls_df = gr.Dataframe(label="MCP tool call trace", wrap=True) | |
| audit_df = gr.Dataframe(label="Audit log", wrap=True) | |
| rca_json = gr.Code(label="Full RCA JSON", language="json") | |
| with gr.Tab("Synthetic Data Lab"): | |
| gr.Markdown("Generate and inspect synthetic telecom operations data for testing the agents.") | |
| platform_filter = gr.Dropdown(choices=all_platforms, value="All", label="Platform filter") | |
| preview_btn = gr.Button("Generate Synthetic Data Preview") | |
| kpi_df = gr.Dataframe(label="KPI records", wrap=True) | |
| alerts_df = gr.Dataframe(label="Alerts and events", wrap=True) | |
| tickets_df = gr.Dataframe(label="Trouble tickets", wrap=True) | |
| changes_df = gr.Dataframe(label="Change records", wrap=True) | |
| with gr.Tab("MCP Tool Registry"): | |
| registry_btn = gr.Button("Show MCP Tool Registry") | |
| registry_df = gr.Dataframe(label="Registered MCP tools", wrap=True) | |
| gr.Markdown(supervisor_architecture_md()) | |
| with gr.Tab("Batch Scenario Testing"): | |
| gr.Markdown("Run all scenarios to test guardrail outcomes and supervisor routing behavior.") | |
| batch_btn = gr.Button("Run Batch Tests") | |
| batch_df = gr.Dataframe(label="Batch test results", wrap=True) | |
| batch_summary = gr.Markdown() | |
| run_btn.click( | |
| fn=run_supervisor_demo, | |
| inputs=[scenario, seed, minutes, execute], | |
| outputs=[cards, kpi_plot, topology_plot, decision_md, agents_df, calls_df, audit_df, rca_json], | |
| ) | |
| preview_btn.click( | |
| fn=generate_data_preview, | |
| inputs=[scenario, seed, minutes, platform_filter], | |
| outputs=[kpi_df, alerts_df, tickets_df, changes_df], | |
| ) | |
| registry_btn.click( | |
| fn=get_registry_preview, | |
| inputs=[scenario, seed], | |
| outputs=[registry_df], | |
| ) | |
| batch_btn.click( | |
| fn=run_batch_tests, | |
| inputs=[seed, execute], | |
| outputs=[batch_df, batch_summary], | |
| ) | |
| demo.load( | |
| fn=run_supervisor_demo, | |
| inputs=[scenario, seed, minutes, execute], | |
| outputs=[cards, kpi_plot, topology_plot, decision_md, agents_df, calls_df, audit_df, rca_json], | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| build_app().launch() | |