How to Integrate LangGraph for payments with Redis for production AI

Combining LangGraph for payments with Redis gives you a practical control plane for payment-aware AI agents. LangGraph handles the workflow state and branching logic, while Redis gives you fast shared memory for session state, idempotency keys, rate limits, and payment event caching.

That pairing is useful when an agent needs to create a payment intent, wait for confirmation, recover from retries, and keep the conversation consistent across workers or replicas.

Prerequisites

• •Python 3.10+
• •A Redis instance running locally or in production
• •langgraph installed
• •redis Python client installed
• •Access to your payment provider SDK or API
• •Environment variables configured:
  • •REDIS_URL
  • •PAYMENTS_API_KEY

Install the dependencies:

pip install langgraph redis requests

Integration Steps

1. •Create a Redis client and define your payment state

Use Redis as the shared store for graph execution metadata and payment status. Keep the state minimal and serializable.

import os
import json
import redis
from typing import TypedDict, Optional

redis_client = redis.from_url(
    os.environ["REDIS_URL"],
    decode_responses=True,
)

class PaymentState(TypedDict):
    user_id: str
    amount_cents: int
    currency: str
    payment_intent_id: Optional[str]
    payment_status: Optional[str]
    conversation_id: str

2. •Build LangGraph nodes for payment creation and status tracking

LangGraph’s StateGraph lets you model the payment flow as explicit steps. In production, that matters because retries and branching need to be deterministic.

import requests
from langgraph.graph import StateGraph, END

PAYMENTS_API_BASE = "https://api.example-payments.com/v1"
PAYMENTS_API_KEY = os.environ["PAYMENTS_API_KEY"]

def create_payment_intent(state: PaymentState) -> PaymentState:
    payload = {
        "amount": state["amount_cents"],
        "currency": state["currency"],
        "metadata": {
            "user_id": state["user_id"],
            "conversation_id": state["conversation_id"],
        },
    }

    resp = requests.post(
        f"{PAYMENTS_API_BASE}/payment_intents",
        headers={"Authorization": f"Bearer {PAYMENTS_API_KEY}"},
        json=payload,
        timeout=15,
    )
    resp.raise_for_status()
    data = resp.json()

    state["payment_intent_id"] = data["id"]
    state["payment_status"] = data["status"]
    return state

def fetch_payment_status(state: PaymentState) -> PaymentState:
    intent_id = state["payment_intent_id"]
    resp = requests.get(
        f"{PAYMENTS_API_BASE}/payment_intents/{intent_id}",
        headers={"Authorization": f"Bearer {PAYMENTS_API_KEY}"},
        timeout=15,
    )
    resp.raise_for_status()
    data = resp.json()

    state["payment_status"] = data["status"]
    return state

graph = StateGraph(PaymentState)
graph.add_node("create_payment", create_payment_intent)
graph.add_node("fetch_status", fetch_payment_status)

graph.set_entry_point("create_payment")
graph.add_edge("create_payment", "fetch_status")
graph.add_edge("fetch_status", END)

app = graph.compile()

3. •Persist execution metadata in Redis

This is where Redis earns its keep. Store a conversation-scoped record so any worker can resume the flow without guessing what happened before.

def save_payment_context(state: PaymentState) -> None:
    key = f"payment:{state['conversation_id']}"
    redis_client.setex(
        key,
        3600,
        json.dumps({
            "user_id": state["user_id"],
            "amount_cents": state["amount_cents"],
            "currency": state["currency"],
            "payment_intent_id": state.get("payment_intent_id"),
            "payment_status": state.get("payment_status"),
        }),
    )

def load_payment_context(conversation_id: str) -> dict | None:
    raw = redis_client.get(f"payment:{conversation_id}")
    return json.loads(raw) if raw else None

4. •Add idempotency so retries do not double-charge

In production AI systems, retries are normal. Use Redis SETNX semantics through set(..., nx=True) to guard against duplicate intent creation.

def acquire_idempotency_lock(conversation_id: str) -> bool:
    return bool(redis_client.set(
        f"lock:payment:{conversation_id}",
        "1",
        nx=True,
        ex=300,
    ))

def release_idempotency_lock(conversation_id: str) -> None:
    redis_client.delete(f"lock:payment:{conversation_id}")

def run_payment_flow(initial_state: PaymentState) -> PaymentState:
    if not acquire_idempotency_lock(initial_state["conversation_id"]):
        cached = load_payment_context(initial_state["conversation_id"])
        if cached:
            return cached  # type: ignore[return-value]
        raise RuntimeError("Payment flow already in progress")

    try:
        result = app.invoke(initial_state)
        save_payment_context(result)
        return result
    finally:
        release_idempotency_lock(initial_state["conversation_id"])

5. •Wire it into your agent entrypoint

Your agent can now call the graph when a user confirms a charge, subscription upgrade, or invoice payment.

if __name__ == "__main__":
    initial_state: PaymentState = {
        "user_id": "user_123",
        "amount_cents": 4999,
        "currency": "usd",
        "payment_intent_id": None,
        "payment_status": None,
        "conversation_id": "conv_abc_001",
    }

    result = run_payment_flow(initial_state)
    print(result)

Testing the Integration

Run a quick smoke test against Redis and your payments endpoint.

test_key = "healthcheck:payments"
redis_client.set(test_key, "ok", ex=30)

value = redis_client.get(test_key)
print("redis:", value)

sample_context = load_payment_context("conv_abc_001")
print("context:", sample_context)

Expected output:

redis: ok
context: {'user_id': 'user_123', 'amount_cents': 4999, 'currency': 'usd', 'payment_intent_id': 'pi_...', 'payment_status': 'succeeded'}

If you want a stricter check, verify the graph returns a terminal status:

result = run_payment_flow({
    "user_id": "user_123",
    "amount_cents": 4999,
    "currency": "usd",
    "payment_intent_id": None,
    "payment_status": None,
    "conversation_id": "conv_abc_002",
})

assert result["payment_status"] in {"requires_action", "processing", "succeeded"}
print("integration passed")

Real-World Use Cases

• •Payment support agents that create intents, poll status, and answer “did my card go through?” without losing context between turns.
• •Subscription workflows where an agent upgrades plans, stores idempotency keys in Redis, and resumes after webhook delays.
• •Fraud-aware assistants that cache risk signals in Redis and branch LangGraph flows based on payment verification results.

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.