Agentic Platform Flowchart
app/main.py
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.api.routes import router
from app.core.config import get_settings
from app.db.session import init_db
from app.middleware.logging import request_logging
from app.middleware.tracing import tracing
@asynccontextmanager
async def lifespan(_:FastAPI):
init_db()
yield
def create_app()-> FastAPI:
settings = get_settings()
logging.basicConfig(level = settings.log_level)
app = FastAPI(title="Service Now Agentic AI", version="1.0.0", lifespan=lifespan)
app.middleware('http')(tracing)
app.middleware('http')(request_logging)
app.include_router(router)
return app
app = create_app()
app/api/routes.py
from uuid import uuid4
from fastapi import APIRouter, Depends
from app.api.dependencies import get_agent_graph, get_memory
from app.core.config import get_settings
from app.middleware.auth import require_user
from app.schemas.chat import ChatRequest, ChatResponse
from app.schemas.incident import IncidentCreate
from app.schemas.response import HealthResponse
from app.tools.incident_tools import create_incident
router = APIRouter()
@router.get("/health", response_model=HealthResponse)
async def health():
settings = get_settings()
return HealthResponse(status="ok", environment=settings.app_env)
@router.post("/v1/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, user: dict = Depends(require_user)):
conversation_id = request.conversation_id or str(uuid4())
memory = get_memory()
memory.add(conversation_id, "user", request.message)
result = await get_agent_graph().ainvoke({"message": request.message, "conversation_id": conversation_id})
memory.add(conversation_id, "assistant", result["answer"])
return ChatResponse(answer=result["answer"], conversation_id=conversation_id, intent=result["intent"], sources=result.get("sources", []))
@router.post("/v1/incidents")
async def open_incident(payload: IncidentCreate, user: dict = Depends(require_user)):
return await create_incident(**payload.model_dump())middleware/auth.py
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from app.services.auth_service import AuthService
bearer = HTTPBearer(auto_error=False)
def require_user(credentials: HTTPAuthorizationCredentials | None = Depends(bearer)) -> dict:
if not credentials: raise HTTPException(status_code=401, detail="Bearer token required")
return AuthService().verify_token(credentials.credentials) what is this auth code is doing explain me line by line in simple way
services/auth_service.py
from datetime import datetime, timedelta, timezone
import jwt
from fastapi import HTTPException, status
from app.core.config import get_settings
class AuthService:
def create_token(self, subject: str, roles: list[str]) -> str:
settings = get_settings()
return jwt.encode(
{
"sub": subject, # ← User ID
"roles": roles, # ← User roles
"iss": settings.jwt_issuer, # ← Issuer (who made this)
"exp": datetime.now(timezone.utc) + timedelta(hours=8) # ← Expires in 8 hours
},
settings.jwt_secret, # ← Secret key for signing
algorithm="HS256" # ← Signing algorithm
)
def verify_token(self, token: str) -> dict:
settings = get_settings()
try:
return jwt.decode(
token, # ← Token to verify
settings.jwt_secret, # ← Secret key to verify
algorithms=["HS256"], # ← Allowed algorithms
issuer=settings.jwt_issuer # ← Check issuer matches
)
except jwt.PyJWTError as exc:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication token"
) from excapp/core/config.py
from functools import lru_cache
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
app_env: str = "development"
log_level: str = "INFO"
gcp_project_id: str = ""
gcp_location: str = "us-central1"
vertex_model: str = "gemini-2.0-flash-001"
servicenow_instance: str = ""
servicenow_username: str = ""
servicenow_password: str = ""
jwt_secret: str = "local-development-secret-change-me"
jwt_issuer: str = "servicenow-agentic-ai"
database_path: str = "./data/agent.db"
enable_write_operations: bool = False
def ensure_data_dir(self) -> None:
Path(self.database_path).parent.mkdir(parents=True, exist_ok=True)
@lru_cache
def get_settings() -> Settings:
settings = Settings()
settings.ensure_data_dir()
return settings
