How to Integrate LangGraph for payments with Redis for multi-agent systems

Combining LangGraph for payments with Redis gives you a clean way to run payment workflows as stateful agent graphs while keeping shared memory, locks, and coordination in Redis. That matters when multiple agents need to check payment status, retry failed steps, or coordinate on the same customer transaction without stepping on each other.

This pattern shows up in billing assistants, refund orchestration, dispute handling, and payment ops tools where one agent creates the workflow and others observe or update state safely.

Prerequisites

• •Python 3.10+
• •A LangGraph-compatible payment workflow package installed
• •Redis 6+ running locally or via managed service
• •redis Python client installed
• •langgraph installed
• •Payment API credentials for your provider
• •Environment variables configured:
  • •REDIS_URL
  • •PAYMENT_API_KEY

Install the core packages:

pip install langgraph redis python-dotenv

Integration Steps

1. •

   Set up Redis as the shared coordination layer

   Use Redis for shared state across agents: locks, checkpoints, task queues, or simple handoff metadata. In multi-agent systems, this prevents duplicate payment actions.

import os
import redis
from dotenv import load_dotenv

load_dotenv()

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

# Basic connectivity check
print(r.ping())

# Shared state for an agent group
r.hset("payment:txn:123", mapping={
    "status": "initiated",
    "owner_agent": "payment_orchestrator",
    "attempts": 0,
})

2. •

   Build a LangGraph payment workflow

   Model the payment process as a graph with explicit nodes for validation, authorization, capture, and reconciliation. LangGraph gives you deterministic control over transitions instead of burying logic inside one long agent loop.

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

class PaymentState(TypedDict):
    txn_id: str
    amount: float
    currency: str
    status: str

def validate_payment(state: PaymentState) -> PaymentState:
    if state["amount"] <= 0:
        return {**state, "status": "rejected"}
    return {**state, "status": "validated"}

def authorize_payment(state: PaymentState) -> PaymentState:
    # Replace with your payment provider SDK call
    return {**state, "status": "authorized"}

def capture_payment(state: PaymentState) -> PaymentState:
    # Replace with your payment provider SDK call
    return {**state, "status": "captured"}

graph = StateGraph(PaymentState)
graph.add_node("validate", validate_payment)
graph.add_node("authorize", authorize_payment)
graph.add_node("capture", capture_payment)

graph.set_entry_point("validate")
graph.add_edge("validate", "authorize")
graph.add_edge("authorize", "capture")
graph.add_edge("capture", END)

payment_app = graph.compile()

3. •

   Persist graph progress in Redis

   Store each step result in Redis so other agents can inspect workflow progress without re-running the graph. This is useful when one agent handles orchestration and another handles customer support or reconciliation.

def save_step(txn_id: str, step: str, payload: dict) -> None:
    key = f"payment:txn:{txn_id}:steps"
    r.hset(key, step, str(payload))

initial_state = {
    "txn_id": "123",
    "amount": 49.99,
    "currency": "USD",
    "status": "pending",
}

result = payment_app.invoke(initial_state)

save_step(result["txn_id"], "final", result)
r.hset(f"payment:txn:{result['txn_id']}", mapping={
    "status": result["status"],
    "last_updated_by": "langgraph_payment_flow",
})

4. •

   Add multi-agent coordination with Redis locks

   When several agents can touch the same transaction, use a Redis lock to serialize critical sections like authorization or capture. This avoids double-charging and race conditions.

lock = r.lock(f"lock:payment:{initial_state['txn_id']}", timeout=30)

if lock.acquire(blocking=True):
    try:
        current_status = r.hget(f"payment:txn:{initial_state['txn_id']}", "status")
        if current_status not in {"captured", "failed"}:
            final_state = payment_app.invoke(initial_state)
            r.hset(f"payment:txn:{final_state['txn_id']}", mapping={
                "status": final_state["status"],
                "processed_by": "agent_a",
            })
    finally:
        lock.release()

5. •

   Wire Redis-backed handoff between agents

   One agent can write a task into Redis after a graph node finishes; another agent can consume it and continue the flow. For production systems, this is cleaner than passing messages through ad hoc in-memory queues.

import json

task = {
    "txn_id": initial_state["txn_id"],
    "next_action": "send_receipt",
    "channel": "email",
}

r.lpush("agent_tasks:payments", json.dumps(task))

queued_task = json.loads(r.rpop("agent_tasks:payments"))
print(queued_task)

Testing the Integration

Run a simple end-to-end test that executes the graph and verifies Redis has the expected state.

test_txn = {
    "txn_id": "test-001",
    "amount": 10.0,
    "currency": "USD",
    "status": "pending",
}

with r.lock("lock:payment:test-001", timeout=10):
    output = payment_app.invoke(test_txn)
    r.hset("payment:txn:test-001", mapping={
        "status": output["status"],
        "currency": output["currency"],
    })

print(output)
print(r.hgetall("payment:txn:test-001"))

Expected output:

{'txn_id': 'test-001', 'amount': 10.0, 'currency': 'USD', 'status': 'captured'}
{'status': 'captured', 'currency': 'USD'}

Real-World Use Cases

• •

  Payment ops assistant
  One agent validates incoming disputes while another checks settlement status in Redis and triggers a LangGraph flow for retries or escalation.

• •

  Refund orchestration
  A coordinator agent writes refund jobs to Redis; worker agents execute LangGraph steps for eligibility checks, approval routing, and refund submission.

• •

  Merchant support automation
  Support agents read transaction state from Redis while a LangGraph workflow handles payment lookup, failure classification, and receipt generation.

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.