File size: 3,060 Bytes
a55d5e3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
# -*- coding: utf-8 -*-
"""main.py
Automatically generated by Colab.
Original file is located at
https://colab.research.google.com/drive/1dfhQA9-peYN1-9q7ueAUF974jYXx3bKQ
"""
from typing import Dict, Any, Optional, List
from fastapi import FastAPI # FastAPI: lightweight, high-performance web framework
from pydantic import BaseModel, Field # Pydantic models enforce input/output schemas
from tools import get_tools # your deterministic toolbelt (modernization, resilience, etc.)
from agent import ( # agent core: LLM brain, memory, prompt, runner
create_agent,
run_infra_resilience_agent,
ReasoningTrace
)
app = FastAPI( # create the app instance
title="InfraResilience Agent API", # helpful metadata (shows in /docs)
version="0.1.0",
description="Modernization + Resilience planning agent (LangChain + MCP + Tools)."
)
# ---------- Request / Response Schemas (Pydantic) ----------
class AnalyzeStackRequest(BaseModel):
"""Input contract for /analyze-stack."""
legacy_stack: Dict[str, Any] = Field(..., description="Legacy stack as a dict (scheduler, language, architecture, dependencies, etc.)")
outage_scenario: Optional[str] = Field(None, description="Optional failure scenario to plan resilience tests (e.g., 'Oracle DB down 30 min').")
class AnalyzeStackResponse(BaseModel):
"""Output contract for /analyze-stack."""
modernization_plan: List[str] # actionable modernization bullets
resilience_strategy: List[str] # resilience drills/test ideas
reasoning_summary: str # short, user-facing reasoning summary
reasoning_trace: List[str] # coarse trace for observability (not raw CoT)
# ---------- Single agent instance (cheap + thread-safe for dev) ----------
TOOLS = get_tools() # build your tool list once at startup
AGENT = create_agent(TOOLS) # wire LLM + tools + memory (MCP)
# ---------- Route: POST /analyze-stack ----------
@app.post("/analyze-stack", response_model=AnalyzeStackResponse)
def analyze_stack(payload: AnalyzeStackRequest) -> AnalyzeStackResponse:
"""
Accepts a legacy stack + optional outage scenario,
invokes the agent, and returns modernization + resilience guidance.
"""
trace = ReasoningTrace() # lightweight trace for clients (observability)
trace.add("API request accepted. Dispatching to agent.")
result = run_infra_resilience_agent( # call your brain with the structured inputs
agent_executor=AGENT,
legacy_stack=payload.legacy_stack,
outage_scenario=payload.outage_scenario,
trace=trace,
)
return AnalyzeStackResponse(**result) # pydantic enforces shape/types on the way out
# ---------- Optional: simple health endpoint ----------
@app.get("/healthz")
def health() -> Dict[str, str]:
"""K8s/ECS-friendly liveness probe."""
return {"status": "ok"} |