How to Integrate LangGraph for payments with Kubernetes for production AI

Combining LangGraph for payments with Kubernetes gives you a clean path to production-grade payment agents: orchestration for payment workflows, and containerized execution with scaling, isolation, and rollout control. The practical win is simple — you can run an AI agent that initiates, validates, retries, and reconciles payment actions while Kubernetes handles reliability, autoscaling, and deployment safety.

Prerequisites

• •Python 3.10+
• •A Kubernetes cluster with kubectl access
• •pip installed
• •A LangGraph account or project with API credentials for your payments workflow
• •A Kubernetes namespace for the agent runtime
• •Basic familiarity with Docker and YAML manifests
• •Environment variables configured:
  • •LANGGRAPH_API_KEY
  • •LANGGRAPH_API_URL
  • •KUBECONFIG or in-cluster service account credentials

Integration Steps

1. •Set up the LangGraph payments client in Python

Start by connecting your agent runtime to the LangGraph payments API. In practice, this means your app can create payment flows, trigger graph runs, and inspect execution state from Python.

import os
from langgraph_sdk import get_client

client = get_client(
    url=os.environ["LANGGRAPH_API_URL"],
    api_key=os.environ["LANGGRAPH_API_KEY"],
)

# Example: list available assistants/graphs in the workspace
assistants = client.assistants.list()
print(assistants)

For production payment flows, keep the graph ID or assistant ID in config. Don’t hardcode it in the app.

2. •Create a payment workflow invocation from your agent

Your agent should call the graph when it needs to authorize or process a payment action. The exact graph structure depends on your implementation, but the SDK pattern stays the same: create a thread, then start a run.

import os
from langgraph_sdk import get_client

client = get_client(
    url=os.environ["LANGGRAPH_API_URL"],
    api_key=os.environ["LANGGRAPH_API_KEY"],
)

thread = client.threads.create()
print(f"thread_id={thread['thread_id']}")

run = client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id=os.environ["LANGGRAPH_ASSISTANT_ID"],
    input={
        "amount": 1250,
        "currency": "USD",
        "merchant_id": "mrc_9012",
        "action": "capture_payment"
    },
)

print(f"run_id={run['run_id']}")

This is the point where your AI system hands off structured payment intent to LangGraph. Keep inputs narrow and validated before they reach the graph.

3. •Add Kubernetes access from the same Python service

Now wire in Kubernetes so the agent can run inside a cluster and optionally inspect its own environment or trigger operational checks. Use the official Kubernetes Python client.

from kubernetes import client as k8s_client
from kubernetes import config as k8s_config

# Use this locally; in-cluster code should use load_incluster_config()
k8s_config.load_kube_config()

v1 = k8s_client.CoreV1Api()
pods = v1.list_namespaced_pod(namespace="payments-agent")

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

In production, this lets you confirm whether the agent pods are healthy before dispatching sensitive workflows like payment capture or refund.

4. •Run LangGraph inside Kubernetes with environment-based configuration

Package the agent as a container and pass configuration through Kubernetes secrets and env vars. Your Python entrypoint can connect to both systems at startup.

import os
from kubernetes import config as k8s_config
from langgraph_sdk import get_client

def build_clients():
    # Works in-cluster when running inside Kubernetes
    try:
        k8s_config.load_incluster_config()
    except Exception:
        k8s_config.load_kube_config()

    lg_client = get_client(
        url=os.environ["LANGGRAPH_API_URL"],
        api_key=os.environ["LANGGRAPH_API_KEY"],
    )
    return lg_client

if __name__ == "__main__":
    client = build_clients()
    assistant_id = os.environ["LANGGRAPH_ASSISTANT_ID"]

    thread = client.threads.create()
    run = client.runs.create(
        thread_id=thread["thread_id"],
        assistant_id=assistant_id,
        input={"action": "refund_payment", "payment_id": "pay_12345", "reason": "duplicate_charge"},
    )

    print(run)

A good pattern here is one pod per worker role:

• •API pod for request intake
• •worker pod for async LangGraph runs
• •reconciler pod for status polling and retry handling

5. •Deploy with a Kubernetes manifest that exposes config cleanly

Use Secrets for credentials and ConfigMaps for non-sensitive settings. Your app reads them as environment variables.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: payments-agent
spec:
  replicas: 2
  selector:
    matchLabels:
      app: payments-agent
  template:
    metadata:
      labels:
        app: payments-agent
    spec:
      containers:
        - name: agent
          image: ghcr.io/your-org/payments-agent:latest
          env:
            - name: LANGGRAPH_API_URL
              valueFrom:
                secretKeyRef:
                  name: langgraph-secrets
                  key: api_url
            - name: LANGGRAPH_API_KEY
              valueFrom:
                secretKeyRef:
                  name: langgraph-secrets
                  key: api_key
            - name: LANGGRAPH_ASSISTANT_ID
              valueFrom:
                configMapKeyRef:
                  name: payments-agent-config
                  key: assistant_id

This gives you repeatable deployments and makes it easy to rotate credentials without rebuilding images.

Testing the Integration

Use a small smoke test that checks both systems from inside the same runtime:

import os
from kubernetes import config as k8s_config, client as k8s_client
from langgraph_sdk import get_client

def test_integration():
    try:
        k8s_config.load_incluster_config()
    except Exception:
        k8s_config.load_kube_config()

    v1 = k8s_client.CoreV1Api()
    ns = os.getenv("POD_NAMESPACE", "payments-agent")
    pods = v1.list_namespaced_pod(namespace=ns)
    assert len(pods.items) > 0, "No pods found"

    lg = get_client(
        url=os.environ["LANGGRAPH_API_URL"],
        api_key=os.environ["LANGGRAPH_API_KEY"],
    )

    thread = lg.threads.create()
    run = lg.runs.create(
        thread_id=thread["thread_id"],
        assistant_id=os.environ["LANGGRAPH_ASSISTANT_ID"],
        input={"action": "health_check", "source": "kubernetes"},
    )

    print("Kubernetes pods:", len(pods.items))
    print("LangGraph run:", run["run_id"])

if __name__ == "__main__":
    test_integration()

Expected output:

Kubernetes pods: 2
LangGraph run: run_01JABCDEF1234567890XYZ

If this passes, your app can see the cluster and successfully invoke a LangGraph workflow.

Real-World Use Cases

• •Payment authorization agents that validate risk signals, call external PSPs, then update workflow state through LangGraph while running on Kubernetes for HA.
• •Refund orchestration systems that queue refund requests, fan out checks across worker pods, and use LangGraph to manage retry logic and manual review branches.
• •Reconciliation agents that poll transaction status from multiple services, compare ledger states, and scale horizontally during end-of-day batch windows.

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.