How to Integrate LangGraph for banking with Kubernetes for AI agents

Combining LangGraph for banking with Kubernetes gives you a clean split between orchestration and execution. LangGraph handles the agent workflow, state, and decisioning; Kubernetes handles scaling, isolation, and lifecycle for the workloads your banking agents trigger.

That matters when you need auditability, retry control, and predictable deployment behavior for sensitive workflows like fraud review, customer servicing, or payment exception handling.

Prerequisites

• •Python 3.10+
• •Access to a Kubernetes cluster
• •kubectl configured against that cluster
• •A container registry your cluster can pull from
• •LangGraph installed:
  • •pip install langgraph
• •Kubernetes Python client installed:
  • •pip install kubernetes
• •Bank-side service credentials available as environment variables or Kubernetes Secrets
• •A deployed banking tool/service your agent can call over HTTP or internal RPC

Integration Steps

1. •Build a LangGraph state machine for the banking workflow.

Start by defining the graph state and the nodes that decide whether to escalate work into Kubernetes. In banking systems, this usually means routing high-risk cases to isolated workers.

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

class BankingState(TypedDict):
    messages: Annotated[list, add_messages]
    case_id: str
    risk_score: float
    k8s_job_name: str | None

def assess_risk(state: BankingState) -> BankingState:
    # Replace with your model or rules engine
    score = 0.92 if "fraud" in state["messages"][-1].content.lower() else 0.15
    return {**state, "risk_score": score}

def route_case(state: BankingState) -> str:
    return "k8s_worker" if state["risk_score"] > 0.8 else "close_case"

graph = StateGraph(BankingState)
graph.add_node("assess_risk", assess_risk)
graph.add_node("k8s_worker", lambda s: s)
graph.add_node("close_case", lambda s: s)

graph.add_edge(START, "assess_risk")
graph.add_conditional_edges("assess_risk", route_case)
graph.add_edge("k8s_worker", END)
graph.add_edge("close_case", END)

app = graph.compile()

2. •Use the Kubernetes Python client to submit a job from inside a graph node.

This is the core integration point. When the graph decides a case needs isolated execution, create a Kubernetes Job using client.BatchV1Api().create_namespaced_job().

from kubernetes import client, config

def submit_k8s_job(state: BankingState) -> BankingState:
    config.load_incluster_config()  # use load_kube_config() locally
    batch_api = client.BatchV1Api()

    job_name = f"banking-agent-{state['case_id']}".lower()

    job_manifest = client.V1Job(
        metadata=client.V1ObjectMeta(name=job_name),
        spec=client.V1JobSpec(
            ttl_seconds_after_finished=300,
            backoff_limit=2,
            template=client.V1PodTemplateSpec(
                metadata=client.V1ObjectMeta(labels={"app": "banking-agent"}),
                spec=client.V1PodSpec(
                    restart_policy="Never",
                    containers=[
                        client.V1Container(
                            name="worker",
                            image="registry.example.com/banking-agent-worker:latest",
                            env=[
                                client.V1EnvVar(name="CASE_ID", value=state["case_id"]),
                                client.V1EnvVar(name="RISK_SCORE", value=str(state["risk_score"])),
                            ],
                        )
                    ],
                ),
            ),
        ),
    )

    batch_api.create_namespaced_job(namespace="banking-ai", body=job_manifest)
    return {**state, "k8s_job_name": job_name}

3. •Wire the Kubernetes submission node into the LangGraph workflow.

Now connect the routing logic to the job submission step. Keep this node small and deterministic so retries do not create duplicate side effects without controls.

def k8s_worker(state: BankingState) -> BankingState:
    # Idempotency check should be backed by Redis/Postgres in production.
    if state.get("k8s_job_name"):
        return state
    return submit_k8s_job(state)

graph = StateGraph(BankingState)
graph.add_node("assess_risk", assess_risk)
graph.add_node("k8s_worker", k8s_worker)
graph.add_node("close_case", lambda s: s)

graph.add_edge(START, "assess_risk")
graph.add_conditional_edges("assess_risk", route_case)
graph.add_edge("k8s_worker", END)
graph.add_edge("close_case", END)

app = graph.compile()

4. •Read back pod/job status and feed it into the agent state.

Banking workflows need traceability. After submitting work to Kubernetes, poll job status using BatchV1Api.read_namespaced_job_status() and update the graph state so downstream nodes can decide whether to notify an analyst or retry.

from kubernetes.client.exceptions import ApiException

def check_job_status(state: BankingState) -> BankingState:
    if not state.get("k8s_job_name"):
        return state

    config.load_incluster_config()
    batch_api = client.BatchV1Api()

    try:
        job = batch_api.read_namespaced_job_status(
            name=state["k8s_job_name"],
            namespace="banking-ai",
        )
        succeeded = job.status.succeeded or 0
        failed = job.status.failed or 0

        return {
            **state,
            "messages": state["messages"] + [
                {"role": "system", "content": f"Job status - succeeded={succeeded}, failed={failed}"}
            ],
        }
    except ApiException as e:
        return {
            **state,
            "messages": state["messages"] + [
                {"role": "system", "content": f"Failed to read job status: {e.reason}"}
            ],
        }

5. •Run the compiled graph from your application entrypoint.

Your application can now accept a banking case event, run it through LangGraph, and dispatch compute to Kubernetes only when needed.

from langchain_core.messages import HumanMessage

initial_state: BankingState = {
    "messages": [HumanMessage(content="Investigate possible fraud on account 1234")],
    "case_id": "case-10021",
    "risk_score": 0.0,
    "k8s_job_name": None,
}

result = app.invoke(initial_state)
print(result)

Testing the Integration

Run a smoke test against a real cluster namespace and verify that the Job gets created.

from kubernetes import client, config

config.load_kube_config()
batch_api = client.BatchV1Api()

job_name = "banking-agent-case-10021"
job = batch_api.read_namespaced_job(name=job_name, namespace="banking-ai")

print(job.metadata.name)
print(job.status.active if job.status else None)

Expected output:

banking-agent-case-10021
1

If you want stronger verification, also confirm pod creation:

core_api = client.CoreV1Api()
pods = core_api.list_namespaced_pod(
    namespace="banking-ai",
    label_selector="app=banking-agent",
)

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

Real-World Use Cases

• •Fraud review agents that route suspicious transactions into isolated Kubernetes jobs for feature enrichment and policy checks.
• •Customer servicing agents that spin up per-case workers for dispute processing, document extraction, and evidence collection.
• •Payments exception handlers that use LangGraph for routing logic while Kubernetes runs retryable background tasks with strict resource limits.

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.