LangGraph StateGraph pipeline that replaced a manual IT access ticket queue at Capgemini. Dual-mode classifier (LLM + Regex fallback), human-in-the-loop approval via interrupt/resume, real-time SSE streaming, and PostgresSaver checkpointing for full audit trails.
In enterprise IT environments, access requests (such as requesting read access to production databases or cloud infrastructure roles) sit in manual ticket queues. Security administrators read each ticket, cross-reference employee credentials, evaluate access scope, and execute provisioning commands manually.
This workflow creates two problems:
This project automates IT access ticket triage and execution using a stateful 3-node LangGraph StateGraph pipeline with explicit Human-In-The-Loop (HITL) authorization gates.
The pipeline consists of three core state nodes:
interrupt_before=["hitl_approval"]. The complete execution state is persisted to PostgreSQL via PostgresSaver.Command(resume="APPROVED")), the agent executes provisioning commands against target identity providers using Model Context Protocol (MCP) tools.Running full LLM inference on every routine ticket adds unnecessary API cost and latency. The classification step uses a fallback pattern:
1import re2from pydantic import BaseModel, Field3
4class AccessIntent(BaseModel):5 user_id: str6 target_system: str7 access_level: str8 confidence: float9
10def classify_ticket_intent(raw_text: str) -> AccessIntent:11 # High-confidence Regex match for standard ticket templates12 pattern = r"Grant\s+(?P<user_id>EMP-\d+)\s+access\s+to\s+(?P<system>[\w-]+)\s+as\s+(?P<level>\w+)"13 match = re.search(pattern, raw_text, re.IGNORECASE)14
15 if match:16 return AccessIntent(17 user_id=match.group("user_id"),18 target_system=match.group("system"),19 access_level=match.group("level"),20 confidence=0.98 # High confidence skips secondary validation21 )22
23 # Fallback to structured LLM classification for unstructured tickets24 return call_llm_classifier(raw_text)For standard structural tickets, the Regex fallback executes in under 2ms with 0.98 confidence, bypassing unnecessary LLM overhead.
Security policies dictate that an AI agent cannot provision production database access autonomously. LangGraph enforces this constraint at the graph compilation layer:
1from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver2from langgraph.graph import StateGraph, START, END3from langgraph.types import Command, interrupt4
5# Compile graph with PostgresSaver checkpointing and interrupt gate6checkpointer = AsyncPostgresSaver(pool)7app = graph.compile(8 checkpointer=checkpointer,9 interrupt_before=["provision_access"]10)11
12# Admin Approval Execution API Endpoint13@router.post("/api/tickets/{thread_id}/approve")14async def approve_ticket(thread_id: str, admin_decision: str):15 config = {"configurable": {"thread_id": thread_id}}16
17 # Resume the suspended graph execution with explicit approval state18 result = await app.ainvoke(19 Command(resume={"status": admin_decision, "admin": "admin@company.com"}),20 config=config21 )22 return {"status": "resumed", "state": result}text/event-stream), enabling security admins to view execution state deltas live without refreshing.PostgresSaver checkpointing ensures 100% state persistence across server restarts or admin review delays.