An End-to-End Guide to OAuth 2.0, Secret Management, and Multi-Agent LangGraph RAG

1. The Use Case: "CompliAgent" - An Enterprise Compliance Assistant

Imagine a Fortune 500 company with 40,000 employees, 12 regulatory jurisdictions, and a policy library of 25,000 documents (HR handbooks, GDPR playbooks, SOX controls, vendor agreements). Employees, managers, and compliance officers need instant, accurate answers like:

"Can I hire a contractor in Germany for 6 months under our current policy, and what approvals are needed?"

A naive RAG chatbot will hallucinate, leak PII, or quote revoked policies. An enterprise-grade system must:

Authenticate & authorize every request (OAuth 2.0 + RBAC).
Protect secrets (API keys, DB credentials, signing keys) via a vault.
Reason across multiple agents (retrieval, analysis, writing, auditing).
Maintain memory & state across turns and sessions.
Produce an auditable trail for regulators.

This article walks through the complete architecture and code for such a system, called CompliAgent.

2. High-Level Architecture

3. Secret Management Strategy

Before a single line of business logic, secrets must be solved correctly. Never commit .env files, never hardcode API keys, and never pass them through environment variables in production containers if you can avoid it.

3.1 The Layered Approach

Layer	Purpose	Tool
L1 — Vault	Runtime secrets (DB creds, LLM API keys, signing keys)	HashiCorp Vault (KV v2 + dynamic Postgres secrets)
L2 — KMS	Envelope encryption for data at rest	AWS KMS / GCP KMS
L3 — HSM	Root key material	AWS CloudHSM / on-prem HSM
L4 — Audit	Every secret access is logged	Vault audit device → SIEM

3.2 Vault Integration Pattern

# secrets/vault_client.py
import hvac
from functools import lru_cache
from pydantic import SecretStr

class VaultClient:
    def __init__(self, url: str, role: str, jwt_path: str):
        # Auth via Kubernetes service account (or AppRole in non-k8s)
        self.client = hvac.Client(url=url)
        with open(jwt_path) as f:
            jwt = f.read()
        self.client.auth.kubernetes.login(role=role, jwt=jwt)

    def get_secret(self, path: str, key: str) -> SecretStr:
        """Fetches a secret; Vault logs this access."""
        data = self.client.secrets.kv.v2.read_secret_version(path=path)
        return SecretStr(data["data"]["data"][key])

@lru_cache(maxsize=1)
def get_vault() -> VaultClient:
    return VaultClient(
        url=os.environ["VAULT_ADDR"],
        role="compliagent-prod",
        jwt_path="/var/run/secrets/kubernetes.io/serviceaccount/token",
    )

def get_llm_api_key() -> SecretStr:
    return get_vault().get_secret("llm/openai", "api_key")

Why This Matters

Dynamic secrets: Vault can generate short-lived Postgres credentials per pod, rotated every hour.
Lease-based: LLM API keys are wrapped with TTLs; if leaked, they expire.
Audit trail: Every get_secret call is logged to Vault's audit device and streamed to your SIEM.

4. OAuth 2.0 Implementation

4.1 The Flows We Support

Flow	Use Case
Authorization Code + PKCE	Web/mobile end-users
Client Credentials	Service-to-service (e.g., audit service calling the LLM gateway)
Resource Owner Password	Never — deprecated in OAuth 2.1

We use Keycloak (or Auth0/Okta) as the IdP. The FastAPI service is a Resource Server that validates JWTs.

4.2 JWT Validation with RBAC

# auth/oauth2.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from pydantic import BaseModel

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

class UserContext(BaseModel):
    sub: str                # user id
    email: str
    roles: list[str]        # ["employee", "manager", "compliance_officer"]
    scopes: list[str]       # ["rag:read", "rag:write", "admin:audit"]
    department: str | None

def decode_token(token: str) -> UserContext:
    try:
        # Fetch JWKS from IdP and cache
        payload = jwt.decode(
            token,
            key=JWKS_PUBLIC_KEY,        # rotated via JWKS endpoint
            algorithms=["RS256"],
            audience="compliagent-api",
            issuer="https://idp.company.com/realms/internal",
        )
        return UserContext(
            sub=payload["sub"],
            email=payload.get("email", ""),
            roles=payload.get("realm_access", {}).get("roles", []),
            scopes=payload.get("scope", "").split(),
            department=payload.get("department"),
        )
    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid token: {e}"
        )

def get_current_user(token: str = Depends(oauth2_scheme)) -> UserContext:
    return decode_token(token)

def require_scope(*required: str):
    def checker(user: UserContext = Depends(get_current_user)):
        missing = set(required) - set(user.scopes)
        if missing:
            raise HTTPException(
                status_code=403,
                detail=f"Missing scopes: {missing}"
            )
        return user
    return checker

4.3 Policy-Based Access Control on RAG

Not every user should see every document. We enforce document-level ACLs by tagging each vector with allowed roles and filtering at retrieval time.

# retrieval/acl_filter.py
def build_acl_filter(user: UserContext) -> dict:
    """pgvector metadata filter: document must be visible to user's roles."""
    return {"allowed_roles": {"$in": user.roles}}

This is combined with the vector similarity search so that a junior employee never retrieves a document tagged ["compliance_officer", "legal"].

5. Multi-Agent LangGraph RAG with Memory & State

5.1 Why LangGraph Over a Single Chain?

A single LangChain RetrievalQA chain cannot:

Route to different tools based on intent.
Recover from retrieval failures.
Produce an audit trail with intermediate reasoning.
Maintain typed state across turns.

LangGraph gives us a state machine where each node is an agent, and the state is a typed dictionary that flows through the graph.

5.2 The State Schema

# agents/state.py
from typing import Annotated, TypedDict, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class CompliState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    user: dict
    intent: Literal[
        "policy_qa",
        "hr_request",
        "legal_review",
        "chitchat"
    ] | None
    retrieved_docs: list[dict]
    analysis: str | None
    response: str | None
    audit_trail: list[dict]
    needs_human_review: bool

5.3 The Agents

# agents/nodes.py
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.prebuilt import create_react_agent

llm = AzureChatOpenAI(
    azure_deployment="gpt-4o",
    api_key=get_llm_api_key().get_secret_value(),
    temperature=0.1,
)

# --- Supervisor / Router ---
def supervisor(state: CompliState) -> CompliState:
    prompt = ChatPromptTemplate.from_messages([
        (
            "system",
            "Classify the user intent into one of: "
            "policy_qa, hr_request, legal_review, chitchat."
        ),
        ("human", "{input}"),
    ])

    intent = (
        prompt
        | llm.with_structured_output({
            "type": "object",
            "properties": {
                "intent": {
                    "type": "string",
                    "enum": [
                        "policy_qa",
                        "hr_request",
                        "legal_review",
                        "chitchat"
                    ]
                }
            }
        })
    ).invoke({"input": state["messages"][-1].content})

    return {
        **state,
        "intent": intent["intent"],
        "audit_trail": state["audit_trail"] + [
            {"node": "supervisor", "intent": intent["intent"]}
        ]
    }

# --- Retriever (RAG with ACL) ---
def retriever(state: CompliState) -> CompliState:
    user = state["user"]
    query = state["messages"][-1].content

    acl_filter = build_acl_filter(UserContext(**user))

    docs = vectorstore.similarity_search(
        query,
        k=6,
        filter=acl_filter
    )

    # Also check document freshness
    docs = [d for d in docs if is_current_version(d)]

    return {
        **state,
        "retrieved_docs": [d.to_dict() for d in docs],
        "audit_trail": state["audit_trail"] + [{
            "node": "retriever",
            "n_docs": len(docs),
            "sources": [
                d.metadata["source"] for d in docs
            ]
        }]
    }

# --- Analyzer (reasoning over docs) ---
def analyzer(state: CompliState) -> CompliState:
    docs_text = "\n---\n".join(
        d["content"] for d in state["retrieved_docs"]
    )

    prompt = ChatPromptTemplate.from_messages([
        (
            "system",
            "You are a compliance analyst. Given the retrieved policy "
            "excerpts, produce a structured analysis. If the excerpts "
            "do not answer the question, say INSUFFICIENT_EVIDENCE."
        ),
        (
            "human",
            "Question: {q}\n\nExcerpts:\n{docs}"
        ),
    ])

    analysis = (prompt | llm).invoke({
        "q": state["messages"][-1].content,
        "docs": docs_text
    })

    return {
        **state,
        "analysis": analysis.content,
        "audit_trail": state["audit_trail"] + [{
            "node": "analyzer",
            "has_evidence":
                "INSUFFICIENT_EVIDENCE" not in analysis.content
        }]
    }

# --- Writer ---
def writer(state: CompliState) -> CompliState:
    prompt = ChatPromptTemplate.from_messages([
        (
            "system",
            "Write a clear, citation-bearing response. "
            "Cite sources as [1], [2]. If uncertain, disclose it."
        ),
        ("human", "Question: {q}\nAnalysis: {a}"),
    ])

    resp = (prompt | llm).invoke({
        "q": state["messages"][-1].content,
        "a": state["analysis"]
    })

    return {**state, "response": resp.content}

# --- Auditor (compliance + PII scrub + log) ---
def auditor(state: CompliState) -> CompliState:
    response = state["response"]

    # PII scrubbing regex + NER model
    scrubbed = scrub_pii(response)

    needs_human = (
        "INSUFFICIENT_EVIDENCE"
        in (state["analysis"] or "")
    )

    persist_audit_log(
        user_id=state["user"]["sub"],
        query=state["messages"][-1].content,
        response=scrubbed,
        trail=state["audit_trail"],
    )

    return {
        **state,
        "response": scrubbed,
        "needs_human_review": needs_human
    }

5.4 The Graph

# agents/graph.py
from langgraph.graph import StateGraph, START, END

def route_after_supervisor(state: CompliState) -> str:
    if state["intent"] == "chitchat":
        return "writer"
    return "retriever"

def route_after_analyzer(state: CompliState) -> str:
    if "INSUFFICIENT_EVIDENCE" in (state["analysis"] or ""):
        return "writer"
    return "writer"

graph = StateGraph(CompliState)

graph.add_node("supervisor", supervisor)
graph.add_node("retriever", retriever)
graph.add_node("analyzer", analyzer)
graph.add_node("writer", writer)
graph.add_node("auditor", auditor)

graph.add_edge(START, "supervisor")

graph.add_conditional_edges(
    "supervisor",
    route_after_supervisor,
    {
        "retriever": "retriever",
        "writer": "writer"
    }
)

graph.add_edge("retriever", "analyzer")

graph.add_conditional_edges(
    "analyzer",
    route_after_analyzer,
    {
        "writer": "writer"
    }
)

graph.add_edge("writer", "auditor")
graph.add_edge("auditor", END)

app = graph.compile(
    checkpointer=RedisSaver.from_url(
        os.environ["REDIS_URL"]
    )
)

The RedisSaver checkpointer means every invocation is keyed by thread_id. The next message from the same user with the same thread_id resumes the exact state — that is your conversation memory.

5.5 Long-Term Memory (User Profile)

Beyond conversation memory, we store a user profile in Postgres: preferred department, past queries, and acknowledged policies. This is injected into the state at START by a tiny pre-node.

def inject_user_profile(state: CompliState) -> CompliState:
    profile = user_profile_repo.get(state["user"]["sub"])

    return {
        **state,
        "user": {
            **state["user"],
            "profile": profile
        }
    }

6. FastAPI Integration

# api/main.py
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from slowapi import Limiter
from slowapi.util import get_remote_address

from auth.oauth2 import (
    get_current_user,
    require_scope,
    UserContext
)

from agents.graph import app as graph_app
from agents.state import CompliState

app = FastAPI(
    title="CompliAgent API",
    version="1.4.0"
)

app.state.limiter = Limiter(
    key_func=get_remote_address
)

@app.post("/v1/chat", response_model=ChatResponse)
async def chat(
    req: ChatRequest,
    user: UserContext = Depends(
        require_scope("rag:read")
    ),
):
    config = {
        "configurable": {
            "thread_id": req.thread_id
        }
    }

    # Build initial state
    initial_state: CompliState = {
        "messages": req.messages,
        "user": user.model_dump(),
        "intent": None,
        "retrieved_docs": [],
        "analysis": None,
        "response": None,
        "audit_trail": [],
        "needs_human_review": False,
    }

    final = await graph_app.ainvoke(
        initial_state,
        config=config
    )

    return ChatResponse(
        response=final["response"],
        citations=[
            d["source"]
            for d in final["retrieved_docs"]
        ],
        needs_human_review=final["needs_human_review"],
        audit_id=persist_audit(
            final["audit_trail"]
        ),
    )

6.1 What This Endpoint Gives You, for Free

Authentication: JWT validated against IdP JWKS.
Authorization: rag:read scope required; admin endpoints require admin:audit.
Rate limiting: per-IP via slowapi.
Stateful multi-turn: same thread_id resumes the LangGraph state.
Audit trail: every node's decision is persisted.
PII scrubbing: done in the auditor node.
Document ACL: enforced at retrieval time.

7. End-to-End Request Flow (Real-Time Walkthrough)

User (a German HR manager) sends:

"Can I hire a contractor in Germany for 6 months under our current policy, and what approvals are needed?"

FastAPI validates the JWT → UserContext(roles=["manager","hr"], department="EMEA").
Scope check passes (rag:read).
Supervisor classifies intent → hr_request.
Retriever runs vector search with filter:

{
  "allowed_roles": {
    "$in": ["manager", "hr"]
  }
}

and country: "DE".

Returns 5 docs:

Global Contractor Policy v4.2
Germany Local Addendum
Additional policy references

Analyzer produces structured analysis:

"Yes, ≤6 months is permitted without works council approval; >6 months requires it. Required forms: HR-DE-101, vendor due-diligence."

Writer emits a response with citations [1], [2].
Auditor scrubs any accidental PII, logs the full trail to the audit table, and flags needs_human_review=false.
Response returned to user. The LangGraph state is checkpointed in Redis under thread_id, so the next question ("What's the form HR-DE-101?") has full context.

8. Production Hardening Checklist

Concern	Mitigation
Prompt injection	System prompt hardening + output classifier in auditor node
Hallucination	INSUFFICIENT_EVIDENCE escape hatch + mandatory citations
Stale policies	Version metadata + is_current_version gate
PII leakage	NER scrubber + regex in auditor
Secrets leak	Vault dynamic secrets + short TTLs + audit device
Token replay	JWT short TTL (5 min) + refresh tokens + rotation
DDoS	Rate limit + WAF + per-user quotas
Observability	OpenTelemetry traces per graph node → Jaeger
Compliance	Audit table immutable (append-only), replicated to cold storage
Failover	LangGraph checkpointer on Redis Cluster; Postgres primary/replica

9. Deployment Sketch

# k8s deployment (abbreviated)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: compliagent

spec:
  template:
    spec:
      serviceAccountName: compliagent-sa

      containers:
      - name: api
        image: compliagent:1.4.0

        env:
        - name: VAULT_ADDR
          value: "https://vault.internal:8200"

        # NO secrets here — fetched at runtime from Vault

        resources:
          requests:
            cpu: "2"
            memory: "4Gi"

          limits:
            cpu: "4"
            memory: "8Gi"

        livenessProbe:
          httpGet:
            path: /healthz
            port: 8000

The pod authenticates to Vault via its Kubernetes service account JWT — no secrets ever touch the manifest or the environment. Build it this way, and you can stand in front of a regulator and show them exactly who asked what, which documents were used, which agent reasoned over them, and why the answer was what it was — with cryptographic proof at every step.

Summary

CompliAgent demonstrates how to build an enterprise-grade compliance assistant that goes far beyond a basic RAG chatbot. By combining OAuth 2.0 authentication, RBAC authorization, Vault-based secret management, ACL-aware retrieval, LangGraph multi-agent orchestration, persistent memory, audit logging, PII protection, and regulatory-grade observability, organizations can deliver accurate, secure, explainable, and compliant AI-powered decision support at enterprise scale.