Enterprise Integration Patterns for LLM Systems

# Example: FastAPI middleware for OIDC token validation in an LLM service
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import httpx
from jose import jwt, JWTError
from functools import lru_cache

app = FastAPI()
bearer_scheme = HTTPBearer()

OIDC_ISSUER = "https://login.enterprise.com/realms/prod"
OIDC_AUDIENCE = "llm-platform"

@lru_cache(maxsize=1)
def get_jwks():
 """Fetch and cache the OIDC provider's public keys."""
 discovery = httpx.get(f"{OIDC_ISSUER}/.well-known/openid-configuration").json()
 return httpx.get(discovery["jwks_uri"]).json()

async def validate_token(
 credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
) -> dict:
 """Validate the Bearer token and return user claims."""
 try:
 claims = jwt.decode(
 credentials.credentials,
 get_jwks(),
 algorithms=["RS256"],
 audience=OIDC_AUDIENCE,
 issuer=OIDC_ISSUER,
 )
 return {
 "user_id": claims["sub"],
 "email": claims.get("email"),
 "roles": claims.get("realm_access", {}).get("roles", []),
 "tenant_id": claims.get("tenant_id"),
 "access_token": credentials.credentials, # for passthrough
 }
 except JWTError as e:
 raise HTTPException(status_code=401, detail=f"Invalid token: {e}")

@app.post("/chat")
async def chat(request: Request, user: dict = Depends(validate_token)):
 """Chat endpoint with identity-aware context."""
 body = await request.json()
 # User identity flows into the LLM context and tool calls
 return await process_chat(
 message=body["message"],
 user_id=user["user_id"],
 tenant_id=user["tenant_id"],
 roles=user["roles"],
 passthrough_token=user["access_token"],
 )

# Example: LLM-specific RBAC policy engine
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class LLMPermission:
 """A single permission for an LLM resource."""
 resource_type: str # "model", "tool", "assistant", "knowledge_base"
 resource_id: str # "gpt-4o", "sql-query", "legal-reviewer", "hr-docs"
 actions: list[str] # ["invoke"], ["read", "write"], ["query"]

@dataclass
class LLMRole:
 """An enterprise role with LLM-specific permissions."""
 name: str
 permissions: list[LLMPermission] = field(default_factory=list)
 token_budget_daily: int = 100_000
 max_context_window: int = 32_000

# Define roles
ROLES = {
 "analyst": LLMRole(
 name="analyst",
 permissions=[
 LLMPermission("model", "gpt-4o-mini", ["invoke"]),
 LLMPermission("tool", "sql-read", ["invoke"]),
 LLMPermission("knowledge_base", "product-docs", ["query"]),
 ],
 token_budget_daily=200_000,
 ),
 "engineer": LLMRole(
 name="engineer",
 permissions=[
 LLMPermission("model", "gpt-4o", ["invoke"]),
 LLMPermission("model", "claude-sonnet-4", ["invoke"]),
 LLMPermission("tool", "sql-read", ["invoke"]),
 LLMPermission("tool", "code-exec", ["invoke"]),
 LLMPermission("knowledge_base", "product-docs", ["query"]),
 LLMPermission("knowledge_base", "internal-wiki", ["query"]),
 ],
 token_budget_daily=500_000,
 ),
 "admin": LLMRole(
 name="admin",
 permissions=[
 LLMPermission("model", "*", ["invoke"]),
 LLMPermission("tool", "*", ["invoke"]),
 LLMPermission("knowledge_base", "*", ["query"]),
 LLMPermission("assistant", "*", ["invoke", "configure"]),
 ],
 token_budget_daily=2_000_000,
 ),
}

def check_permission(
 user_roles: list[str],
 resource_type: str,
 resource_id: str,
 action: str,
) -> bool:
 """Check if any of the user's roles grant the requested permission."""
 for role_name in user_roles:
 role = ROLES.get(role_name)
 if role is None:
 continue
 for perm in role.permissions:
 if perm.resource_type != resource_type:
 continue
 if perm.resource_id != "*" and perm.resource_id != resource_id:
 continue
 if action in perm.actions:
 return True
 return False

# Example: Tenant-isolated RAG pipeline with namespace separation
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

class TenantIsolatedRAG:
 """RAG pipeline with strict tenant data isolation."""

 def __init__(self, qdrant_url: str):
 self.client = QdrantClient(url=qdrant_url)

 def retrieve(
 self,
 tenant_id: str,
 query_embedding: list[float],
 top_k: int = 5,
 ) -> list[dict]:
 """Retrieve documents scoped to a single tenant."""
 results = self.client.search(
 collection_name="documents",
 query_vector=query_embedding,
 query_filter=Filter(
 must=[
 FieldCondition(
 key="tenant_id",
 match=MatchValue(value=tenant_id),
 )
 ]
 ),
 limit=top_k,
 )
 return [
 {
 "text": hit.payload["text"],
 "source": hit.payload["source"],
 "score": hit.score,
 }
 for hit in results
 ]

 def build_context(
 self,
 tenant_id: str,
 query: str,
 query_embedding: list[float],
 system_prompt_override: str | None = None,
 ) -> list[dict]:
 """Build a tenant-scoped context for the LLM."""
 docs = self.retrieve(tenant_id, query_embedding)
 # Use tenant-specific system prompt if configured
 system_prompt = (
 system_prompt_override
 or self.get_tenant_system_prompt(tenant_id)
 )
 context_block = "\n\n".join(
 f"[Source: {d['source']}]\n{d['text']}" for d in docs
 )
 return [
 {"role": "system", "content": system_prompt},
 {"role": "user", "content": f"Context:\n{context_block}\n\nQuestion: {query}"},
 ]

 def get_tenant_system_prompt(self, tenant_id: str) -> str:
 """Retrieve the tenant-specific system prompt from configuration."""
 # In production, this reads from a database or config service
 return f"You are an assistant for tenant {tenant_id}. Only use provided context."

# Example: Structured audit logging for LLM interactions
import json
import hashlib
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class LLMAuditRecord:
 """Immutable audit record for a single LLM interaction."""
 # Identity
 request_id: str
 user_id: str
 tenant_id: str
 session_id: str
 # Request
 timestamp: str
 model: str
 prompt_hash: str # SHA-256 of the full prompt (for privacy)
 prompt_text: Optional[str] # included only if policy allows
 tool_calls: list[dict]
 # Response
 response_hash: str
 response_text: Optional[str]
 # Metrics
 input_tokens: int
 output_tokens: int
 latency_ms: int
 estimated_cost_usd: float
 # Integrity
 record_hash: str = ""

 def compute_hash(self) -> str:
 """Compute a SHA-256 hash of the record for tamper detection."""
 data = asdict(self)
 data.pop("record_hash")
 canonical = json.dumps(data, sort_keys=True)
 return hashlib.sha256(canonical.encode()).hexdigest()

class AuditLogger:
 """Writes LLM audit records to an append-only store."""

 def __init__(self, storage_backend):
 self.storage = storage_backend

 def log(self, record: LLMAuditRecord):
 record.record_hash = record.compute_hash()
 self.storage.append(asdict(record))

 def verify(self, record_dict: dict) -> bool:
 """Verify that a stored record has not been tampered with."""
 stored_hash = record_dict.pop("record_hash")
 canonical = json.dumps(record_dict, sort_keys=True)
 computed = hashlib.sha256(canonical.encode()).hexdigest()
 return computed == stored_hash

# Example: Risk-based approval routing for LLM tool calls
from enum import Enum
from dataclasses import dataclass

class RiskLevel(Enum):
 LOW = "low" # auto-execute
 MEDIUM = "medium" # log and execute, flag for review
 HIGH = "high" # require human approval before execution
 CRITICAL = "critical" # require two approvals

# Risk classification for tool calls
TOOL_RISK_MAP = {
 "search_documents": RiskLevel.LOW,
 "summarize_text": RiskLevel.LOW,
 "send_email": RiskLevel.HIGH,
 "update_database": RiskLevel.HIGH,
 "delete_record": RiskLevel.CRITICAL,
 "approve_expense": RiskLevel.CRITICAL,
 "generate_contract": RiskLevel.HIGH,
 "query_database_readonly": RiskLevel.MEDIUM,
}

@dataclass
class PendingApproval:
 request_id: str
 user_id: str
 tool_name: str
 tool_params: dict
 risk_level: RiskLevel
 llm_reasoning: str
 original_query: str
 status: str = "pending" # pending, approved, rejected, modified

class ApprovalRouter:
 """Routes tool calls based on risk classification."""

 def __init__(self, approval_queue, executor):
 self.queue = approval_queue
 self.executor = executor

 async def route(self, tool_call: dict, context: dict) -> dict:
 tool_name = tool_call["name"]
 risk = TOOL_RISK_MAP.get(tool_name, RiskLevel.HIGH)

 if risk == RiskLevel.LOW:
 return await self.executor.execute(tool_call)

 if risk == RiskLevel.MEDIUM:
 result = await self.executor.execute(tool_call)
 await self.queue.log_for_review(tool_call, context, result)
 return result

 # HIGH and CRITICAL: require approval before execution
 approval = PendingApproval(
 request_id=context["request_id"],
 user_id=context["user_id"],
 tool_name=tool_name,
 tool_params=tool_call["arguments"],
 risk_level=risk,
 llm_reasoning=context.get("reasoning", ""),
 original_query=context["original_query"],
 )
 await self.queue.submit(approval)
 return {
 "status": "pending_approval",
 "message": f"Action '{tool_name}' requires approval. Request submitted.",
 "approval_id": approval.request_id,
 }

# Example: Governance configuration for an enterprise LLM platform
governance:
 data_residency:
 default_region: "eu-west-1"
 allowed_regions: ["eu-west-1", "eu-central-1"]
 blocked_regions: ["us-*", "ap-*"]
 enforcement: "strict" # reject requests that would route outside allowed regions

 model_provenance:
 require_version_pinning: true
 allowed_providers: ["azure-openai", "anthropic-eu", "self-hosted"]
 model_registry_url: "https://models.internal.enterprise.com/registry"
 audit_every_invocation: true

 prompt_management:
 repository: "git@github.com:enterprise/llm-prompts.git"
 require_review: true
 min_reviewers: 2
 deploy_strategy: "canary" # canary, blue-green, or immediate
 canary_percentage: 10
 rollback_on_quality_drop: true
 quality_threshold:
 min_relevance_score: 0.85
 max_hallucination_rate: 0.02

 secrets:
 vault_backend: "hashicorp-vault"
 vault_url: "https://vault.internal.enterprise.com"
 api_key_rotation_days: 30
 tool_credential_path: "secret/llm-platform/tools/"
 audit_secret_access: true

# Example: Vault-integrated credential manager for LLM tool calls
import time
from typing import Optional

class VaultCredentialManager:
 """Manages API credentials with vault integration and caching."""

 def __init__(self, vault_client, cache_ttl_seconds: int = 300):
 self.vault = vault_client
 self.cache_ttl = cache_ttl_seconds
 self._cache: dict[str, tuple[str, float]] = {}

 def get_credential(self, path: str) -> str:
 """Fetch a credential, using cache if fresh."""
 cached = self._cache.get(path)
 if cached:
 value, fetched_at = cached
 if time.time() - fetched_at < self.cache_ttl:
 return value

 # Fetch from vault
 secret = self.vault.secrets.kv.v2.read_secret_version(path=path)
 value = secret["data"]["data"]["api_key"]
 self._cache[path] = (value, time.time())
 return value

 def get_tool_headers(self, tool_name: str) -> dict:
 """Build authentication headers for a specific tool."""
 credential_path = f"llm-platform/tools/{tool_name}"
 api_key = self.get_credential(credential_path)
 return {"Authorization": f"Bearer {api_key}"}

 def rotate_and_verify(self, path: str) -> bool:
 """Force rotation and verify the new credential works."""
 # Invalidate cache
 self._cache.pop(path, None)
 # Fetch new credential (vault handles the actual rotation)
 new_key = self.get_credential(path)
 return new_key is not None

# Complete key lifecycle manager using HashiCorp Vault
import hvac
import secrets
import logging
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

class KeyLifecycleManager:
 """Manages the full lifecycle of API keys via HashiCorp Vault."""

 def __init__(self, vault_url: str, vault_token: str, mount: str = "secret"):
 self.client = hvac.Client(url=vault_url, token=vault_token)
 self.mount = mount

 def generate(self, path: str, key_length: int = 64) -> str:
 """Phase 1: Generate a cryptographically strong key and store it."""
 new_key = secrets.token_urlsafe(key_length)
 self.client.secrets.kv.v2.create_or_update_secret(
 path=path,
 secret=dict(
 api_key=new_key,
 created_at=datetime.now(timezone.utc).isoformat(),
 status="active",
 ),
 mount_point=self.mount,
 )
 logger.info("Generated new key at path=%s", path)
 return new_key

 def rotate(self, path: str) -> str:
 """Phase 3: Rotate by generating a new key, keeping the old version."""
 old = self.client.secrets.kv.v2.read_secret_version(
 path=path, mount_point=self.mount,
 )
 old_version = old["data"]["metadata"]["version"]
 new_key = self.generate(path)
 logger.info(
 "Rotated key at path=%s (old version=%d)", path, old_version
 )
 return new_key

 def revoke(self, path: str) -> None:
 """Phase 4: Mark the current key as revoked."""
 self.client.secrets.kv.v2.create_or_update_secret(
 path=path,
 secret=dict(
 api_key="REVOKED",
 revoked_at=datetime.now(timezone.utc).isoformat(),
 status="revoked",
 ),
 mount_point=self.mount,
 )
 logger.warning("Revoked key at path=%s", path)

 def destroy(self, path: str, versions: list[int]) -> None:
 """Phase 5: Permanently destroy specific key versions."""
 self.client.secrets.kv.v2.destroy_secret_versions(
 path=path, versions=versions, mount_point=self.mount,
 )
 logger.warning(
 "Destroyed versions=%s at path=%s", versions, path
 )

# Example: Enterprise chatbot orchestrator combining all integration patterns
class EnterpriseChatOrchestrator:
 """
 Production chatbot with SSO, RBAC, tenant isolation,
 audit logging, approval workflows, and vault integration.
 """

 def __init__(
 self,
 rag: TenantIsolatedRAG,
 audit: AuditLogger,
 approval: ApprovalRouter,
 credentials: VaultCredentialManager,
 ):
 self.rag = rag
 self.audit = audit
 self.approval = approval
 self.credentials = credentials

 async def handle_request(self, user: dict, message: str) -> dict:
 """Process a chat request through the full enterprise pipeline."""
 request_id = generate_request_id()

 # 1. RBAC check: can this user invoke the requested model?
 model = self.select_model(user["roles"])
 if not check_permission(user["roles"], "model", model, "invoke"):
 return {"error": "Insufficient permissions for model access."}

 # 2. Tenant-isolated retrieval
 embedding = await self.embed(message)
 context_messages = self.rag.build_context(
 tenant_id=user["tenant_id"],
 query=message,
 query_embedding=embedding,
 )

 # 3. LLM invocation with tool routing
 api_key = self.credentials.get_credential(f"models/{model}")
 response = await self.call_llm(model, context_messages, api_key)

 # 4. Process tool calls through approval workflow
 if response.get("tool_calls"):
 for tool_call in response["tool_calls"]:
 if not check_permission(
 user["roles"], "tool", tool_call["name"], "invoke"
 ):
 continue # skip unauthorized tools
 result = await self.approval.route(
 tool_call,
 context={
 "request_id": request_id,
 "user_id": user["user_id"],
 "original_query": message,
 },
 )
 # Feed tool result back to LLM for final response

 # 5. Audit logging
 self.audit.log(LLMAuditRecord(
 request_id=request_id,
 user_id=user["user_id"],
 tenant_id=user["tenant_id"],
 session_id=user.get("session_id", ""),
 timestamp=datetime.now(timezone.utc).isoformat(),
 model=model,
 prompt_hash=hashlib.sha256(message.encode()).hexdigest(),
 prompt_text=None, # redacted per policy
 tool_calls=response.get("tool_calls", []),
 response_hash=hashlib.sha256(
 response["content"].encode()
 ).hexdigest(),
 response_text=response["content"],
 input_tokens=response["usage"]["input"],
 output_tokens=response["usage"]["output"],
 latency_ms=response["latency_ms"],
 estimated_cost_usd=response["estimated_cost"],
 ))

 return {"response": response["content"], "request_id": request_id}

 def select_model(self, roles: list[str]) -> str:
 """Select the best available model based on user roles."""
 if "admin" in roles or "engineer" in roles:
 return "gpt-4o"
 return "gpt-4o-mini"