How to Integrate LangGraph for pension funds with Kubernetes for production AI

Combining LangGraph for pension funds with Kubernetes gives you a clean path from agent logic to production-grade execution. LangGraph handles the stateful workflow and decisioning for pension-fund operations, while Kubernetes gives you scaling, rollout control, and isolation for workloads that must survive load spikes and node failures.

This matters when you’re building agents for pension administration, member support, contribution reconciliation, or document-heavy workflows. You want the graph to stay deterministic and auditable, and you want the runtime to be disposable, observable, and easy to redeploy.

Prerequisites

• •Python 3.10+
• •A Kubernetes cluster:
  • •local: kind, minikube, or k3d
  • •production: EKS, GKE, AKS, or on-prem
• •kubectl configured against your cluster
• •Access to a LangGraph-compatible Python environment
• •Required Python packages:
  • •langgraph
  • •kubernetes
  • •fastapi
  • •uvicorn
• •A container registry for pushing your agent image
• •Basic RBAC permissions in Kubernetes:
  • •create deployments
  • •create services
  • •read pods/logs

Install the Python dependencies:

pip install langgraph kubernetes fastapi uvicorn

Integration Steps

1) Define the LangGraph workflow for the pension-fund agent

Start with a small graph that routes pension queries through validation and decision steps. For production systems, keep the state explicit so every transition is traceable.

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

def merge_messages(left: list[str], right: list[str]) -> list[str]:
    return left + right

class PensionState(TypedDict):
    messages: Annotated[list[str], merge_messages]
    customer_id: str
    request_type: str
    risk_flag: bool

def validate_member(state: PensionState) -> PensionState:
    if not state["customer_id"]:
        raise ValueError("Missing customer_id")
    state["messages"].append("member validated")
    return state

def assess_request(state: PensionState) -> PensionState:
    state["risk_flag"] = state["request_type"] in {"benefit_change", "withdrawal"}
    state["messages"].append(f"risk={state['risk_flag']}")
    return state

graph = StateGraph(PensionState)
graph.add_node("validate_member", validate_member)
graph.add_node("assess_request", assess_request)
graph.set_entry_point("validate_member")
graph.add_edge("validate_member", "assess_request")
graph.add_edge("assess_request", END)

app = graph.compile()

The important part here is that compile() gives you an executable graph object. In production, that object becomes the core business logic you package into a container.

2) Wrap the graph behind an API service

Expose the graph over HTTP so Kubernetes can run it as a stateless service. This keeps your orchestration layer separate from your workflow logic.

from fastapi import FastAPI
from pydantic import BaseModel

api = FastAPI()

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

@api.post("/run")
def run_graph(req: PensionRequest):
    initial_state = {
        "messages": [],
        "customer_id": req.customer_id,
        "request_type": req.request_type,
        "risk_flag": False,
    }
    result = app.invoke(initial_state)
    return result

Run it locally first:

uvicorn main:api --host 0.0.0.0 --port 8000

At this point you have a clean boundary:

• •LangGraph owns workflow execution.
• •FastAPI exposes a stable contract.
• •Kubernetes will later own scheduling and resilience.

3) Containerize the service for Kubernetes

Package the service into an image so it can be deployed anywhere.

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY main.py .

EXPOSE 8000
CMD ["uvicorn", "main:api", "--host", "0.0.0.0", "--port", "8000"]

Use a minimal requirements.txt:

langgraph
kubernetes
fastapi
uvicorn

Build and push:

docker build -t registry.example.com/pension-agent:1.0.0 .
docker push registry.example.com/pension-agent:1.0.0

4) Deploy to Kubernetes with a Service and Deployment

Use Kubernetes to manage replicas and expose the agent internally or externally.

from kubernetes import client, config

config.load_kube_config()

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

deployment = client.V1Deployment(
    metadata=client.V1ObjectMeta(name="pension-agent"),
    spec=client.V1DeploymentSpec(
        replicas=2,
        selector=client.V1LabelSelector(match_labels={"app": "pension-agent"}),
        template=client.V1PodTemplateSpec(
            metadata=client.V1ObjectMeta(labels={"app": "pension-agent"}),
            spec=client.V1PodSpec(
                containers=[
                    client.V1Container(
                        name="agent",
                        image="registry.example.com/pension-agent:1.0.0",
                        ports=[client.V1ContainerPort(container_port=8000)],
                    )
                ]
            ),
        ),
    ),
)

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

Create the service:

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

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

This is where Kubernetes adds value:

• •replica management for load spikes
• •self-healing if pods crash
• •rollout control when you ship graph changes

5) Trigger workflows from another pod or internal job

In real systems, another service or batch job usually calls your agent endpoint. Use standard Python HTTP calls from inside the cluster or from an internal worker.

import requests

response = requests.post(
    "http://pension-agent.default.svc.cluster.local/run",
    json={
        "customer_id": "MEM-10291",
        "request_type": "benefit_change",
    },
    timeout=10,
)

print(response.status_code)
print(response.json())

If you need direct Kubernetes interaction from Python—for example to scale during peak processing—you can use the same SDK:

from kubernetes import client, config

config.load_kube_config()
apps_v1 = client.AppsV1Api()

scale = client.V1Scale(
    spec=client.V1ScaleSpec(replicas=4)
)

apps_v1.replace_namespaced_deployment_scale(
    name="pension-agent",
    namespace="default",
    body=scale,
)

Testing the Integration

Run this smoke test after deployment:

import requests

payload = {
    "customer_id": "MEM-10291",
    "request_type": "withdrawal"
}

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

Expected output:

200
{'messages': ['member validated', 'risk=True'], 'customer_id': 'MEM-10291', 'request_type': 'withdrawal', 'risk_flag': True}

If that passes, verify Kubernetes sees healthy pods:

kubectl get pods -l app=pension-agent
kubectl get svc pension-agent
kubectl logs deploy/pension-agent

Real-World Use Cases

• •

  Member request triage

  • •Route benefit changes, withdrawals, and contribution disputes through different graph branches.
  • •Scale replicas in Kubernetes during month-end spikes.
• •

  Document processing pipelines

  • •Use LangGraph nodes to extract data from pension forms.
  • •Run OCR workers and validation services as separate Kubernetes deployments.
• •

  Compliance review agents

  • •Build audit-friendly approval flows for high-risk pension actions.
  • •Keep each step isolated in containers with logs collected by cluster tooling.

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.