How to Integrate LangGraph for payments with Redis for startups

Combining LangGraph for payments with Redis gives you a practical pattern for payment-aware agents: stateful workflows, retry-safe execution, and fast shared memory across requests. For startups, that means you can build agents that create invoices, track payment status, pause on risk checks, and resume when webhooks arrive without losing context.

Prerequisites

• •Python 3.10+
• •A Redis instance running locally or via Redis Cloud
• •A LangGraph setup with your agent workflow defined
• •Access to your payment provider API keys
• •Installed packages:
  • •langgraph
  • •redis
  • •python-dotenv
  • •your payment SDK, for example stripe

Install dependencies:

pip install langgraph redis python-dotenv stripe

Set environment variables:

export REDIS_URL="redis://localhost:6379/0"
export STRIPE_API_KEY="sk_test_..."

Integration Steps

1. •Create a Redis client for shared state and idempotency

Use Redis to store workflow checkpoints, payment session metadata, and idempotency keys. In startup systems, this is what keeps your agent from double-charging when a request retries.

import os
import redis

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
r = redis.Redis.from_url(REDIS_URL, decode_responses=True)

# Smoke test connection
print(r.ping())

A common pattern is to namespace payment data by conversation or order ID:

order_id = "order_123"
r.hset(f"payments:{order_id}", mapping={
    "status": "pending",
    "customer_id": "cus_001",
    "amount": "4999",
    "currency": "usd",
})

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

LangGraph works best when your graph state is explicit. Define the fields your payment agent needs to move through the workflow.

from typing import TypedDict, Optional

class PaymentState(TypedDict):
    order_id: str
    customer_id: str
    amount: int
    currency: str
    payment_intent_id: Optional[str]
    status: str

This state gets passed between nodes like create_payment, confirm_payment, and sync_redis.

3. •Define LangGraph nodes that call your payment provider and persist status in Redis

Here’s a simple Stripe-backed node function. It creates a PaymentIntent and writes the result into Redis so another worker or webhook handler can pick it up later.

import os
import stripe

stripe.api_key = os.getenv("STRIPE_API_KEY")

def create_payment(state: PaymentState) -> PaymentState:
    intent = stripe.PaymentIntent.create(
        amount=state["amount"],
        currency=state["currency"],
        metadata={
            "order_id": state["order_id"],
            "customer_id": state["customer_id"],
        },
    )

    r.hset(
        f"payments:{state['order_id']}",
        mapping={
            "payment_intent_id": intent["id"],
            "status": intent["status"],
        },
    )

    return {
        **state,
        "payment_intent_id": intent["id"],
        "status": intent["status"],
    }

Now wire that into a LangGraph workflow using StateGraph:

from langgraph.graph import StateGraph, END

workflow = StateGraph(PaymentState)
workflow.add_node("create_payment", create_payment)
workflow.set_entry_point("create_payment")
workflow.add_edge("create_payment", END)

app = workflow.compile()

4. •Use Redis as the coordination layer for retries and webhook updates

For production payments, you usually don’t finish everything in one request. Store the current status in Redis and update it from webhook events or background workers.

def mark_paid(order_id: str) -> None:
    r.hset(f"payments:{order_id}", mapping={"status": "paid"})

def get_payment_status(order_id: str) -> dict:
    return r.hgetall(f"payments:{order_id}")

If you want LangGraph to resume from stored state, load the latest values from Redis before invoking the graph:

def load_state(order_id: str) -> PaymentState:
    data = r.hgetall(f"payments:{order_id}")
    return {
        "order_id": order_id,
        "customer_id": data.get("customer_id", ""),
        "amount": int(data.get("amount", "0")),
        "currency": data.get("currency", "usd"),
        "payment_intent_id": data.get("payment_intent_id"),
        "status": data.get("status", "new"),
    }

5. •Invoke the graph and persist the final result

This is where the agent execution starts. The graph creates the payment intent, then Redis keeps the canonical record.

initial_state = {
    "order_id": "order_123",
    "customer_id": "cus_001",
    "amount": 4999,
    "currency": "usd",
    "payment_intent_id": None,
    "status": "new",
}

result = app.invoke(initial_state)
print(result)
print(r.hgetall("payments:order_123"))

Testing the Integration

Run this end-to-end check against test mode:

test_order = {
    "order_id": "order_test_001",
    "customer_id": "cus_test_001",
    "amount": 1500,
    "currency": "usd",
    "payment_intent_id": None,
    "status": "new",
}

result = app.invoke(test_order)
stored = r.hgetall("payments:order_test_001")

print("LangGraph result:", result)
print("Redis record:", stored)
assert stored["payment_intent_id"] == result["payment_intent_id"]
assert stored["status"] == result["status"]

Expected output:

LangGraph result: {'order_id': 'order_test_001', 'customer_id': 'cus_test_001', 'amount': 1500, 'currency': 'usd', 'payment_intent_id': 'pi_...', 'status': 'requires_payment_method'}
Redis record: {'payment_intent_id': 'pi_...', 'status': 'requires_payment_method'}

Real-World Use Cases

• •Payment follow-up agents that create an invoice, wait for webhook confirmation in Redis, then notify Slack or email once paid.
• •Retry-safe checkout automation where Redis stores idempotency keys so LangGraph doesn’t create duplicate charges on retries.
• •Collections and dunning workflows that route failed payments through different LangGraph branches based on status stored in Redis.

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.