How to Integrate LangGraph for fintech with Kubernetes for multi-agent systems

Combining LangGraph for fintech with Kubernetes gives you a clean way to run multi-agent financial workflows as managed, observable, and scalable services. In practice, that means you can route customer onboarding, fraud checks, KYC review, and transaction monitoring through separate agents while Kubernetes handles deployment, scaling, restarts, and service discovery.

Prerequisites

• •Python 3.10+
• •A running Kubernetes cluster
  • •local: kind, minikube, or k3d
  • •production: EKS, GKE, or AKS
• •kubectl configured against your cluster
• •A container registry you can push to
• •LangGraph installed:
  • •pip install langgraph
• •Kubernetes Python client installed:
  • •pip install kubernetes
• •Access to your LLM provider credentials in environment variables
• •Basic understanding of:
  • •LangGraph nodes/edges/state
  • •Kubernetes Deployments, Services, ConfigMaps, Secrets

Integration Steps

1. •

   Build a LangGraph workflow for a fintech use case

   Start with a small graph that models one business flow. For fintech, a good first target is transaction review: classify the request, run fraud checks, then decide whether to approve or escalate.

   from typing import TypedDict, Literal
   from langgraph.graph import StateGraph, END
   
   class TxState(TypedDict):
       transaction_id: str
       amount: float
       country: str
       risk_score: int
       decision: str
   
   def classify_tx(state: TxState) -> TxState:
       amount = state["amount"]
       country = state["country"]
   
       risk = 10
       if amount > 5000:
           risk += 40
       if country not in {"US", "GB", "CA"}:
           risk += 25
   
       state["risk_score"] = risk
       return state
   
   def decide(state: TxState) -> TxState:
       state["decision"] = "approve" if state["risk_score"] < 50 else "escalate"
       return state
   
   graph = StateGraph(TxState)
   graph.add_node("classify_tx", classify_tx)
   graph.add_node("decide", decide)
   
   graph.set_entry_point("classify_tx")
   graph.add_edge("classify_tx", "decide")
   graph.add_edge("decide", END)
   
   app = graph.compile()
   

2. •

   Wrap the graph in an API service

   Kubernetes needs something to schedule. The simplest pattern is a FastAPI service that exposes the compiled LangGraph app as an HTTP endpoint.

   from fastapi import FastAPI
   from pydantic import BaseModel
   from langgraph.graph import StateGraph, END
   
   app_api = FastAPI()
   
   class TxRequest(BaseModel):
       transaction_id: str
       amount: float
       country: str
   
   @app_api.post("/review")
   def review_tx(req: TxRequest):
       result = app.invoke({
           "transaction_id": req.transaction_id,
           "amount": req.amount,
           "country": req.country,
           "risk_score": 0,
           "decision": ""
       })
       return result
   

   This is the boundary you want in production: LangGraph handles orchestration inside the process, while Kubernetes handles lifecycle outside it.

3. •

   Add Kubernetes client calls for agent coordination

   In multi-agent systems, one agent often needs to inspect cluster state or trigger another workload. The Kubernetes Python client gives you direct access to pods, jobs, and services.

   from kubernetes import client, config
   
   config.load_incluster_config()  # use load_kube_config() locally
   
   v1 = client.CoreV1Api()
   batch_v1 = client.BatchV1Api()
   
   pods = v1.list_namespaced_pod(namespace="fintech-agents")
   print(f"Running pods: {[p.metadata.name for p in pods.items]}")
   
   job_manifest = client.V1Job(
       metadata=client.V1ObjectMeta(name="fraud-check-job"),
       spec=client.V1JobSpec(
           template=client.V1PodTemplateSpec(
               metadata=client.V1ObjectMeta(labels={"app": "fraud-check"}),
               spec=client.V1PodSpec(
                   restart_policy="Never",
                   containers=[
                       client.V1Container(
                           name="fraud-check",
                           image="registry.example.com/fraud-check:latest",
                           args=["python", "worker.py"]
                       )
                   ]
               )
           )
       )
   )
   
   batch_v1.create_namespaced_job(namespace="fintech-agents", body=job_manifest)
   

4. •

   Connect LangGraph routing to Kubernetes-backed agents

   Use the graph to route work based on risk. Low-risk transactions stay local; high-risk ones trigger a Kubernetes Job or call a separate agent service running in the cluster.

   from kubernetes import client, config
   
   config.load_incluster_config()
   batch_v1 = client.BatchV1Api()
   
   def escalate_to_fraud_agent(state):
       if state["decision"] == "escalate":
           job_name = f"fraud-review-{state['transaction_id']}"
           job = client.V1Job(
               metadata=client.V1ObjectMeta(name=job_name),
               spec=client.V1JobSpec(
                   template=client.V1PodTemplateSpec(
                       metadata=client.V1ObjectMeta(labels={"job": job_name}),
                       spec=client.V1PodSpec(
                           restart_policy="Never",
                           containers=[
                               client.V1Container(
                                   name="fraud-agent",
                                   image="registry.example.com/fraud-agent:latest",
                                   env=[
                                       client.V1EnvVar(name="TX_ID", value=state["transaction_id"])
                                   ]
                               )
                           ]
                       )
                   )
               )
           )
           batch_v1.create_namespaced_job(namespace="fintech-agents", body=job)
       return state
   

5. •

   Deploy both components on Kubernetes

   Package your API container and deploy it as a Service so other agents can call it reliably. Keep secrets out of the image and mount them as environment variables or Kubernetes Secrets.

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: tx-review-api
     namespace: fintech-agents
     labels:
       app: tx-review-api
   spec:
     replicas: 2
     selector:
       matchLabels:
         app: tx-review-api
     template:
       metadata:
         labels:
           app: tx-review-api
       spec:
         containers:
         - name: api
           image: registry.example.com/tx-review-api:latest
           ports:
           - containerPort: 8000
           env:
           - name: OPENAI_API_KEY
             valueFrom:
               secretKeyRef:
                 name: llm-secrets
                 key: openai_api_key
   

apiVersion: v1 kind: Service metadata: name: tx-review-api-svc namespace: fintech-agents spec: selector: app: tx-review-api ports:

• •port: 80 targetPort: 8000 type: ClusterIP

## Testing the Integration

Run the API locally or through the cluster service, then submit a sample transaction and verify that the graph produces a decision and that escalation triggers K8s activity when needed.

```python
import requests

payload = {
    "transaction_id": "tx_10001",
    "amount": 12000,
    "country": "NG"
}

resp = requests.post("http://localhost:8000/review", json=payload)
print(resp.status_code)
print(resp.json())

Expected output:

200
{
  "transaction_id": "tx_10001",
  "amount": 12000,
  "country": "NG",
  "risk_score": 75,
  "decision": "escalate"
}

If you want to verify the Kubernetes side too:

from kubernetes import client, config

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

pods = v1.list_namespaced_pod(namespace="fintech-agents")
print([p.metadata.name for p in pods.items])

You should see your API pod plus any worker pods or Jobs created by the escalation path.

Real-World Use Cases

• •

  Fraud triage pipeline

  • •One agent scores risk.
  • •Another agent fetches customer history.
  • •A Kubernetes Job runs deep inspection only when thresholds are crossed.
• •

  KYC onboarding workflow

  • •LangGraph routes identity checks across document parsing, sanctions screening, and manual review.
  • •Kubernetes scales each stage independently during onboarding spikes.
• •

  Claims or payments operations

  • •One agent validates inputs.
  • •Another agent checks policy rules.
  • •A third agent runs exception handling as an isolated pod when cases need human review.

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.