EAP – How Build Agentic Workflow?
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())app/core/config.py
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 settingsapp/api/dependencies.py
from functools import lru_cache
from app.graph.graph import build_graph
from app.memory.conversation_memory import ConversationMemory
@lru_cache
def get_agent_graph(): return build_graph()
@lru_cache
def get_memory(): return ConversationMemory()app/graph/router.py
def route_intent(state: dict) -> str:
return state.get("intent", "general")app/graph/graph.py
from langgraph.graph import END, START, StateGraph
from app.graph.state import AgentState
from app.graph.nodes import classify_node, incident_node, knowledge_node, request_node, general_node
from app.graph.router import route_intent
def build_graph():
graph = StateGraph(AgentState)
graph.add_node("classify", classify_node)
graph.add_node("incident", incident_node)
graph.add_node("knowledge", knowledge_node)
graph.add_node("request", request_node)
graph.add_node("general", general_node)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {"incident": "incident", "knowledge": "knowledge", "request": "request", "general": "general"})
for node in ("incident", "knowledge", "request", "general"):
graph.add_edge(node, END)
return graph.compile()app/graph/node.py
from app.agents.supervisor_agent import SupervisorAgent
from app.agents.incident_agent import IncidentAgent
from app.agents.knowledge_agent import KnowledgeAgent
from app.agents.request_agent import RequestAgent
from app.services.llm_service import LLMService
async def classify_node(state: dict) -> dict:
return {"intent": SupervisorAgent().route(state["message"])}
async def incident_node(state: dict) -> dict:
answer, sources = await IncidentAgent().run(state["message"])
return {"answer": answer, "sources": sources}
async def knowledge_node(state: dict) -> dict:
answer, sources = await KnowledgeAgent().run(state["message"])
return {"answer": answer, "sources": sources}
async def request_node(state: dict) -> dict:
answer, sources = await RequestAgent().run(state["message"])
return {"answer": answer, "sources": sources}
async def general_node(state: dict) -> dict:
return {"answer": LLMService().generate(f"You are a helpful ServiceNow assistant. Answer the user safely and concisely.\nUser: {state['message']}"), "sources": []}app/services/llm_service.py
import logging
from google import genai
from app.core.config import get_settings
logger = logging.getLogger(__name__)
class LLMService:
"""Vertex AI Gemini wrapper with a deterministic local fallback."""
def __init__(self) -> None:
self.settings = get_settings()
self.client = None
if self.settings.gcp_project_id:
self.client = genai.Client(vertexai=True, project=self.settings.gcp_project_id, location=self.settings.gcp_location)
def generate(self, prompt: str) -> str:
if not self.client:
logger.warning("Vertex AI is not configured; returning local fallback")
return "I need Vertex AI configuration to generate a detailed response."
response = self.client.models.generate_content(model=self.settings.vertex_model, contents=prompt)
return response.text or "I could not generate a response."
def classify(self, message: str) -> str:
lowered = message.lower()
if any(word in lowered for word in ("incident", "outage", "error", "broken", "issue", "ticket")):
return "incident"
if any(word in lowered for word in ("knowledge", "how do", "how to", "documentation", "article")):
return "knowledge"
if any(word in lowered for word in ("request", "catalog", "laptop", "access", "provision")):
return "request"
if self.client:
label = self.generate(f"Classify into incident, knowledge, request, general. Return one label only. Request: {message}").strip().lower()
return label if label in {"incident", "knowledge", "request", "general"} else "general"
return "general"app/agents/supervisor_agent.py
from app.services.llm_service import LLMService
class SupervisorAgent:
def __init__(self): self.llm = LLMService()
def route(self, message: str) -> str: return self.llm.classify(message)app/agents/incident_agent.py
from app.core.prompts import INCIDENT_PROMPT
from app.services.llm_service import LLMService
from app.tools.incident_tools import find_incidents
class IncidentAgent:
async def run(self, message: str) -> tuple[str, list[dict]]:
records = await find_incidents(message)
context = str(records) if records else "No matching incident records found."
return LLMService().generate(INCIDENT_PROMPT.format(message=message, context=context)), recordsapp/agents/knowledge_agent.py
from app.core.prompts import KNOWLEDGE_PROMPT
from app.services.llm_service import LLMService
from app.tools.kb_tools import search_knowledge
class KnowledgeAgent:
async def run(self, message: str) -> tuple[str, list[dict]]:
records = await search_knowledge(message)
context = str(records) if records else "No knowledge articles matched."
return LLMService().generate(KNOWLEDGE_PROMPT.format(message=message, context=context)), recordsapp/agents/request_agent.py
from app.core.prompts import REQUEST_PROMPT
from app.services.llm_service import LLMService
from app.tools.request_tools import find_requests
class RequestAgent:
async def run(self, message: str) -> tuple[str, list[dict]]:
records = await find_requests(message)
context = str(records) if records else "No matching request records found."
return LLMService().generate(REQUEST_PROMPT.format(message=message, context=context)), recordsapp/services/servicenow_service.py
from typing import Any
import httpx
from app.core.config import get_settings
class ServiceNowService:
def __init__(self) -> None:
self.settings = get_settings()
@property
def configured(self) -> bool:
return bool(self.settings.servicenow_instance and self.settings.servicenow_username and self.settings.servicenow_password)
async def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
if not self.configured:
return {"result": [], "warning": "ServiceNow is not configured"}
url = f"{self.settings.servicenow_instance.rstrip('/')}/api/now/{path.lstrip('/')}"
async with httpx.AsyncClient(timeout=25) as client:
response = await client.request(method, url, auth=(self.settings.servicenow_username, self.settings.servicenow_password), headers={"Accept": "application/json", "Content-Type": "application/json"}, **kwargs)
response.raise_for_status()
return response.json()
async def query_table(self, table: str, query: str, fields: str, limit: int = 10) -> list[dict]:
data = await self._request("GET", f"table/{table}", params={"sysparm_query": query, "sysparm_fields": fields, "sysparm_limit": limit})
return data.get("result", [])
async def create_incident(self, payload: dict) -> dict:
if not self.settings.enable_write_operations:
return {"warning": "Write operations are disabled. Set ENABLE_WRITE_OPERATIONS=true after approval."}
data = await self._request("POST", "table/incident", json=payload)
return data.get("result", data)app/tools/incident_tools.py
import re
from app.services.servicenow_service import ServiceNowService
service = ServiceNowService()
async def find_incidents(message: str) -> list[dict]:
number = re.search(r"\bINC\d+\b", message.upper())
query = f"number={number.group(0)}" if number else f"short_descriptionLIKE{message[:80]}"
return await service.query_table("incident", query, "number,short_description,state,priority,assigned_to,sys_id")
async def create_incident(short_description: str, description: str, urgency: str = "3", impact: str = "3") -> dict:
return await service.create_incident({"short_description": short_description, "description": description, "urgency": urgency, "impact": impact})app/tools/kb_tools.py
from app.services.servicenow_service import ServiceNowService
service = ServiceNowService()
async def search_knowledge(query: str) -> list[dict]:
return await service.query_table("kb_knowledge", f"short_descriptionLIKE{query[:100]}^ORtextLIKE{query[:100]}", "number,short_description,text,sys_id", 5)app/tools/request_tools.py
import re
from app.services.servicenow_service import ServiceNowService
service = ServiceNowService()
async def find_requests(message: str) -> list[dict]:
number = re.search(r"\bREQ\d+\b", message.upper())
query = f"number={number.group(0)}" if number else f"short_descriptionLIKE{message[:80]}"
return await service.query_table("sc_request", query, "number,short_description,request_state,requested_for,sys_id")
