How to Integrate LangGraph for retail banking with Kubernetes for multi-agent systems

Combining LangGraph with Kubernetes gives you a clean way to run retail banking agents as stateful workflows while keeping the infrastructure elastic, isolated, and observable. The practical win is simple: you can route customer servicing, fraud review, KYC checks, and escalation flows through LangGraph, then scale those agent workers on Kubernetes without rewriting the orchestration layer.

Prerequisites

• •Python 3.10+
• •A Kubernetes cluster with kubectl access
• •kubernetes Python client installed
• •langgraph installed
• •Access to your LLM provider credentials via environment variables
• •A container registry for pushing your agent image
• •Basic familiarity with:
  • •LangGraph StateGraph
  • •Kubernetes Deployments, Services, and ConfigMaps
  • •REST or gRPC endpoints for agent execution

Install the Python dependencies:

pip install langgraph kubernetes pydantic

Integration Steps

1. •Define the retail banking workflow in LangGraph

Start by modeling the banking flow as a state machine. For retail banking, this usually means intake, policy checks, risk scoring, and handoff to a human queue if needed.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

def merge_notes(existing: str, new: str) -> str:
    return existing + "\n" + new if existing else new

class BankingState(TypedDict):
    customer_id: str
    request_type: str
    risk_score: int
    notes: Annotated[str, merge_notes]

def intake_node(state: BankingState) -> BankingState:
    return {**state, "notes": "Intake completed"}

def risk_node(state: BankingState) -> BankingState:
    score = 82 if state["request_type"] == "wire_transfer" else 25
    return {**state, "risk_score": score, "notes": f"Risk scored at {score}"}

def route_node(state: BankingState) -> BankingState:
    return {**state, "notes": "Routed for review"}

graph = StateGraph(BankingState)
graph.add_node("intake", intake_node)
graph.add_node("risk", risk_node)
graph.add_node("route", route_node)

graph.add_edge(START, "intake")
graph.add_edge("intake", "risk")
graph.add_conditional_edges(
    "risk",
    lambda s: "route" if s["risk_score"] > 50 else END,
    {"route": "route", END: END},
)
graph.add_edge("route", END)

app = graph.compile()

2. •Package the graph as a service that Kubernetes can run

Expose the compiled graph through a small API service. In production this is what your pods will execute.

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

app_api = FastAPI()

class RequestPayload(BaseModel):
    customer_id: str
    request_type: str

@app_api.post("/run")
def run_workflow(payload: RequestPayload):
    result = app.invoke({
        "customer_id": payload.customer_id,
        "request_type": payload.request_type,
        "risk_score": 0,
        "notes": ""
    })
    return result

if __name__ == "__main__":
    uvicorn.run(app_api, host="0.0.0.0", port=8080)

Build a container image from this service and push it to your registry.

3. •Deploy the LangGraph service on Kubernetes

Use the Kubernetes Python client when you want to automate deployment from CI/CD or an operator. This example creates a Deployment and Service for your agent runtime.

from kubernetes import client, config

config.load_kube_config()

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

deployment = client.V1Deployment(
    metadata=client.V1ObjectMeta(name="banking-agent"),
    spec=client.V1DeploymentSpec(
        replicas=2,
        selector=client.V1LabelSelector(match_labels={"app": "banking-agent"}),
        template=client.V1PodTemplateSpec(
            metadata=client.V1ObjectMeta(labels={"app": "banking-agent"}),
            spec=client.V1PodSpec(containers=[
                client.V1Container(
                    name="agent",
                    image="registry.example.com/banking-agent:latest",
                    ports=[client.V1ContainerPort(container_port=8080)],
                    env=[
                        client.V1EnvVar(name="OPENAI_API_KEY", value_from=None)
                    ]
                )
            ])
        )
    )
)

apps_v1.create_namespaced_deployment(namespace="default", body=deployment)

service = client.V1Service(
    metadata=client.V1ObjectMeta(name="banking-agent-svc"),
    spec=client.V1ServiceSpec(
        selector={"app": "banking-agent"},
        ports=[client.V1ServicePort(port=80, target_port=8080)],
        type="ClusterIP"
    )
)

core_v1.create_namespaced_service(namespace="default", body=service)

4. •Wire multi-agent routing through Kubernetes services

In multi-agent systems, one graph can call another service based on business rules. For example, a fraud review agent and a customer support agent can be separate deployments behind Services.

import requests

def fraud_review_node(state):
    response = requests.post(
        "http://fraud-agent-svc.default.svc.cluster.local/run",
        json={
            "customer_id": state["customer_id"],
            "request_type": state["request_type"]
        },
        timeout=10,
    )
    response.raise_for_status()
    fraud_result = response.json()
    return {
        **state,
        "notes": state["notes"] + f"\nFraud agent verdict: {fraud_result.get('verdict', 'unknown')}"
    }

Then attach that node into your graph where escalation is required.

5. •Add config and rollout controls through Kubernetes primitives

Keep policy thresholds and routing rules outside code so ops can change them without redeploying the whole stack.

from kubernetes import client

config_map = client.V1ConfigMap(
    metadata=client.V1ObjectMeta(name="banking-agent-config"),
    data={
        "risk_threshold": "50",
        "human_review_queue": "retail-banking-review"
    }
)

# In practice mount this ConfigMap into the pod or inject it as env vars.
# Then read it inside your LangGraph nodes:
import os

RISK_THRESHOLD = int(os.getenv("RISK_THRESHOLD", config_map.data["risk_threshold"]))

This pattern matters in banking because policy changes are frequent and deployment windows are not always aligned with business changes.

Testing the Integration

Run the service locally or against your cluster endpoint, then verify that the workflow executes and that Kubernetes can reach the pod-backed service.

import requests

payload = {
    "customer_id": "CUST-10021",
    "request_type": "wire_transfer"
}

resp = requests.post("http://localhost:8080/run", json=payload, timeout=10)
resp.raise_for_status()
print(resp.json())

Expected output:

{
  "customer_id": "CUST-10021",
  "request_type": "wire_transfer",
  "risk_score": 82,
  "notes": "\nIntake completed\nRisk scored at 82\nRouted for review"
}

If you’re testing Kubernetes connectivity directly:

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()
pods = v1.list_namespaced_pod(namespace="default", label_selector="app=banking-agent")
print([p.metadata.name for p in pods.items])

Real-World Use Cases

• •

  Fraud triage pipeline
  Route high-risk payment requests from LangGraph into a dedicated fraud agent deployment on Kubernetes.

• •

  KYC document processing
  Split document extraction, identity verification, and sanctions screening into separate agents that scale independently.

• •

  Customer servicing orchestration
  Use one graph for intent classification and another for product eligibility checks across multiple services in the cluster.

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.