How to Integrate LangGraph for payments with Redis for AI agents

Combining LangGraph for payments with Redis gives you a clean way to build agent workflows that can charge, track, and resume payment-related actions without losing state. LangGraph handles the orchestration and tool execution, while Redis gives you fast persistence for session state, idempotency keys, and payment event tracking.

Prerequisites

• •Python 3.10+
• •A Redis instance running locally or via Redis Cloud
• •LangGraph installed in your project
• •Access to your payment provider credentials if your graph calls a real payments API
• •pip and a virtual environment
• •Basic familiarity with LangGraph state graphs and Redis key/value operations

Install the dependencies:

pip install langgraph redis pydantic

If you are using LangGraph’s persistence helpers, make sure your version includes checkpoint support.

Integration Steps

1. •Set up Redis as the shared state layer

Use Redis to store workflow state, retry markers, and payment references. In production, this is what keeps your agent from double-charging when a node gets retried.

import redis
import json

redis_client = redis.Redis(
    host="localhost",
    port=6379,
    db=0,
    decode_responses=True,
)

payment_session_id = "pay_sess_123"

redis_client.hset(
    f"payment:{payment_session_id}",
    mapping={
        "status": "pending",
        "amount": "49.99",
        "currency": "USD",
        "user_id": "user_42",
    },
)

print(redis_client.hgetall(f"payment:{payment_session_id}"))

2. •Create a LangGraph state model for the payment flow

Define the state your agent will carry through the graph. Keep it explicit: amount, currency, customer reference, and payment status.

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END

class PaymentState(TypedDict):
    user_id: str
    amount: float
    currency: str
    payment_id: Optional[str]
    status: str

3. •Add nodes that read from and write to Redis

Each node should update Redis after a meaningful transition. That gives you durable visibility into where the payment flow is at any moment.

import uuid

def create_payment_intent(state: PaymentState):
    payment_id = f"pi_{uuid.uuid4().hex[:12]}"

    redis_client.hset(
        f"payment:{payment_id}",
        mapping={
            "user_id": state["user_id"],
            "amount": str(state["amount"]),
            "currency": state["currency"],
            "status": "created",
        },
    )

    return {
        **state,
        "payment_id": payment_id,
        "status": "created",
    }

def mark_payment_confirmed(state: PaymentState):
    redis_client.hset(
        f"payment:{state['payment_id']}",
        mapping={"status": "confirmed"},
    )

    return {
        **state,
        "status": "confirmed",
    }

4. •Wire the nodes into a LangGraph workflow

LangGraph’s StateGraph lets you define the exact path of execution. For a simple payment flow, create intent first, then confirm it.

workflow = StateGraph(PaymentState)

workflow.add_node("create_payment_intent", create_payment_intent)
workflow.add_node("mark_payment_confirmed", mark_payment_confirmed)

workflow.add_edge(START, "create_payment_intent")
workflow.add_edge("create_payment_intent", "mark_payment_confirmed")
workflow.add_edge("mark_payment_confirmed", END)

app = workflow.compile()

If you need persistence across runs, compile with a checkpointer backed by Redis-compatible storage in your stack. The key point is that the graph should be resumable after failures.

5. •Run the graph and persist the final result

Invoke the compiled graph with initial state. Then store the final status in Redis so downstream services can consume it without calling the agent again.

initial_state: PaymentState = {
    "user_id": "user_42",
    "amount": 49.99,
    "currency": "USD",
    "payment_id": None,
    "status": "initiated",
}

result = app.invoke(initial_state)

redis_client.set(
    f"payment_result:{result['payment_id']}",
    json.dumps(result),
)

print(result)

Testing the Integration

Run a quick smoke test to confirm both systems are connected and state is being written correctly.

test_state: PaymentState = {
    "user_id": "user_99",
    "amount": 19.95,
    "currency": "USD",
    "payment_id": None,
    "status": "initiated",
}

result = app.invoke(test_state)
stored = redis_client.hgetall(f"payment:{result['payment_id']}")

print("graph_result:", result)
print("redis_record:", stored)

Expected output:

graph_result: {'user_id': 'user_99', 'amount': 19.95, 'currency': 'USD', 'payment_id': 'pi_ab12cd34ef56', 'status': 'confirmed'}
redis_record: {'user_id': 'user_99', 'amount': '19.95', 'currency': 'USD', 'status': 'confirmed'}

Real-World Use Cases

• •

  Payment approval agents

  • •Route an invoice through review steps, persist each decision in Redis, and let LangGraph resume after human approval.
• •

  Subscription billing assistants

  • •Track renewal attempts, retry windows, and customer notifications without duplicating charges.
• •

  Dispute handling workflows

  • •Store evidence references in Redis while LangGraph coordinates refund checks, fraud rules, and escalation paths.

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.