""" 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 # ----------------------------------------------------------------------------- @dataclass 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 # ----------------------------------------------------------------------------- @dataclass 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]] @dataclass 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 @staticmethod 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 # ----------------------------------------------------------------------------- @dataclass 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"""
{label}
{value}
""" 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"""
Scenario
{result.scenario_name}
Platform
{result.platform}
{result.domain}
Confidence
{result.confidence}%
Evidence-based RCA
Guardrail Risk
{result.risk_level}
Score {result.risk_score}/100
Ticket
{result.ticket_id}
Simulated ticketing MCP
""" 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"""

{APP_TITLE}

{APP_SUBTITLE}

""" ) 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()