| from __future__ import annotations |
|
|
| from fastapi import FastAPI, HTTPException |
| from pydantic import BaseModel |
|
|
| from .mcp import ChangeType, Mode, SimulateRequest, SimulateResponse, simulate_network_change |
|
|
| app = FastAPI(title="Network Change Simulator MCP Server") |
|
|
|
|
| class ToolDescription(BaseModel): |
| name: str |
| description: str |
| input_schema: dict |
| output_schema: dict |
|
|
|
|
| def tool_metadata() -> ToolDescription: |
| input_schema = { |
| "type": "object", |
| "properties": { |
| "change_type": {"type": "string", "enum": [c.value for c in ChangeType]}, |
| "preset_id": {"type": "string"}, |
| "mode": {"type": "string", "enum": [m.value for m in Mode], "default": "lightning"}, |
| }, |
| "required": ["change_type", "preset_id"], |
| } |
| output_schema = { |
| "type": "object", |
| "properties": { |
| "risk_score": {"type": "integer", "minimum": 0, "maximum": 100}, |
| "risk_level": {"type": "string", "enum": ["low", "medium", "high"]}, |
| "pre_check_summary": {"type": "array", "items": {"type": "string"}}, |
| "post_check_summary": {"type": "array", "items": {"type": "string"}}, |
| "blast_radius_summary": {"type": "string"}, |
| "explanation": {"type": "string"}, |
| "mode": {"type": "string"}, |
| "mode_note": {"type": "string"}, |
| }, |
| } |
| return ToolDescription( |
| name="simulate_network_change", |
| description="Simulate lightning or full change risk for VLAN/interface/BGP scenarios.", |
| input_schema=input_schema, |
| output_schema=output_schema, |
| ) |
|
|
|
|
| @app.get("/health") |
| def health() -> dict: |
| return {"status": "ok"} |
|
|
|
|
| @app.get("/mcp/tools") |
| def list_tools() -> dict: |
| tool = tool_metadata() |
| return {"tools": [tool]} |
|
|
|
|
| @app.post("/mcp/simulate", response_model=SimulateResponse) |
| def simulate(request: SimulateRequest) -> SimulateResponse: |
| try: |
| return simulate_network_change(request) |
| except ValueError as exc: |
| raise HTTPException(status_code=400, detail=str(exc)) from exc |
|
|
