How to Integrate LangGraph for banking with Kubernetes for production AI

Banks don’t need another chatbot demo. They need agent systems that can hold state, route work across services, and survive node failures without dropping a customer session.

That’s where LangGraph for banking and Kubernetes fit together. LangGraph gives you the stateful orchestration for things like KYC review, fraud triage, and loan exception handling, while Kubernetes gives you the runtime guarantees to run those agents as production services.

Prerequisites

• •Python 3.10+
• •A running Kubernetes cluster
  • •kubectl configured
  • •access to a namespace for AI workloads
• •A bank-safe LangGraph setup
  • •langgraph
  • •your banking tools wrapped as callable nodes
• •Container registry access for your agent image
• •Secrets configured in Kubernetes
  • •API keys
  • •database credentials
  • •model provider credentials if needed

Install the core packages:

pip install langgraph kubernetes pydantic

Integration Steps

1. •Build the LangGraph banking workflow.

Start by modeling the banking process as a graph. For example, a payment exception flow can route between validation, risk scoring, and escalation.

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class BankingState(TypedDict):
    customer_id: str
    transaction_id: str
    amount: float
    risk_score: int
    decision: Literal["approve", "review", "reject"]

def validate_transaction(state: BankingState):
    if state["amount"] <= 0:
        return {"decision": "reject"}
    return state

def score_risk(state: BankingState):
    score = 85 if state["amount"] > 10000 else 20
    return {"risk_score": score}

def route_decision(state: BankingState):
    if state.get("risk_score", 0) >= 80:
        return {"decision": "review"}
    return {"decision": "approve"}

graph = StateGraph(BankingState)
graph.add_node("validate_transaction", validate_transaction)
graph.add_node("score_risk", score_risk)
graph.add_node("route_decision", route_decision)

graph.set_entry_point("validate_transaction")
graph.add_edge("validate_transaction", "score_risk")
graph.add_edge("score_risk", "route_decision")
graph.add_conditional_edges(
    "route_decision",
    lambda s: s["decision"],
    {
        "approve": END,
        "review": END,
        "reject": END,
    },
)

app = graph.compile()

2. •Wrap the graph in a service boundary that Kubernetes can run.

In production, don’t call the graph directly from a notebook. Expose it through an API process so pods can scale horizontally and be restarted safely.

from fastapi import FastAPI
from pydantic import BaseModel

class TransactionRequest(BaseModel):
    customer_id: str
    transaction_id: str
    amount: float

api = FastAPI()

@api.post("/evaluate")
def evaluate(req: TransactionRequest):
    initial_state = {
        "customer_id": req.customer_id,
        "transaction_id": req.transaction_id,
        "amount": req.amount,
        "risk_score": 0,
        "decision": "approve",
    }
    result = app.invoke(initial_state)
    return result

3. •Package the service for Kubernetes deployment.

Kubernetes needs a container image with your graph runtime and API server. Keep config out of code and inject it through environment variables or secrets.

import os

MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "openai")
BANK_ENV = os.getenv("BANK_ENV", "prod")

def get_runtime_config():
    return {
        "model_provider": MODEL_PROVIDER,
        "bank_env": BANK_ENV,
    }

A minimal deployment manifest should point at that API container and set replica count for availability.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: banking-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: banking-agent
  template:
    metadata:
      labels:
        app: banking-agent
    spec:
      containers:
        - name: api
          image: registry.example.com/banking-agent:v1
          ports:
            - containerPort: 8000
          envFrom:
            - secretRef:
                name: banking-agent-secrets
---
apiVersion: v1
kind: Service
metadata:
  name: banking-agent-svc
spec:
  selector:
    app: banking-agent
  ports:
    - port: 80
      targetPort: 8000

4. •Use the Kubernetes Python client to manage rollout checks from your control plane.

If you’re automating deployments from an internal platform, use the official client to verify that your agent pods are healthy before sending traffic.

from kubernetes import client, config

config.load_kube_config()

apps_v1 = client.AppsV1Api()
pods_v1 = client.CoreV1Api()

deployment = apps_v1.read_namespaced_deployment(
    name="banking-agent",
    namespace="ai-prod",
)

pods = pods_v1.list_namespaced_pod(
    namespace="ai-prod",
    label_selector="app=banking-agent",
)

print("replicas:", deployment.status.replicas)
print("ready_replicas:", deployment.status.ready_replicas)
for pod in pods.items:
    print(pod.metadata.name, pod.status.phase)

5. •Add resilience around retries and idempotency.

Banking workflows need deterministic retries. Keep transaction IDs stable so reruns don’t double-process events when Kubernetes restarts a pod mid-request.

processed_ids = set()

def idempotent_evaluate(transaction_id: str, payload: dict):
    if transaction_id in processed_ids:
        return {"status": "duplicate"}
    
    processed_ids.add(transaction_id)
    return app.invoke(payload)

Testing the Integration

Run the API locally or in-cluster, then verify both the LangGraph execution path and Kubernetes rollout status.

test_payload = {
    "customer_id": "CUST-1001",
    "transaction_id": "TXN-9009",
    "amount": 25000.0,
}

result = app.invoke({
    **test_payload,
    "risk_score": 0,
    "decision": "approve",
})

print(result)

Expected output:

{'customer_id': 'CUST-1001', 'transaction_id': 'TXN-9009', 'amount': 25000.0, 'risk_score': 85, 'decision': 'review'}

If you want to verify Kubernetes too:

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

pods = v1.list_namespaced_pod(namespace="ai-prod", label_selector="app=banking-agent")
assert len(pods.items) > 0

for pod in pods.items:
    print(pod.metadata.name, pod.status.phase)

Real-World Use Cases

• •

  Fraud triage agents
  Route suspicious card transactions through scoring, policy checks, and manual review queues.

• •

  KYC exception handling
  Orchestrate document validation, sanctions screening, and compliance escalation with durable state.

• •

  Loan operations assistants
  Handle missing documents, underwriting exceptions, and customer follow-ups across multiple backend systems.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.