How to Integrate LangChain for payments with PostgreSQL for AI agents

Combining LangChain for payments with PostgreSQL gives you a clean pattern for building AI agents that can both reason about payment actions and persist state reliably. The practical win is simple: the agent can decide when to charge, refund, or verify a payment, while PostgreSQL stores transaction history, audit trails, customer context, and idempotency keys.

Prerequisites

• •Python 3.10+
• •A PostgreSQL instance running locally or in the cloud
• •A LangChain-compatible payments provider account and API key
• •pip installed
• •Access to an AI model provider supported by LangChain
• •Basic familiarity with SQL and Python async/sync database access

Install the core packages:

pip install langchain langchain-openai psycopg2-binary sqlalchemy python-dotenv

If your payments integration uses a dedicated LangChain payment tool package, install that too. The exact package name depends on the provider, but the integration pattern stays the same: wrap payment actions as tools, then persist every action in PostgreSQL.

Integration Steps

1. •Set up your environment variables.

Keep credentials out of code. Store your model key, database URL, and payment API key in .env.

from dotenv import load_dotenv
import os

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
DATABASE_URL = os.getenv("DATABASE_URL")
PAYMENTS_API_KEY = os.getenv("PAYMENTS_API_KEY")

A typical DATABASE_URL looks like this:

postgresql+psycopg2://postgres:password@localhost:5432/agentdb

2. •Create a PostgreSQL connection and audit table.

Use SQLAlchemy for portability. For agent systems, you want a table that records every payment decision and result.

from sqlalchemy import create_engine, text

engine = create_engine(DATABASE_URL)

with engine.begin() as conn:
    conn.execute(text("""
        CREATE TABLE IF NOT EXISTS payment_events (
            id SERIAL PRIMARY KEY,
            user_id TEXT NOT NULL,
            action TEXT NOT NULL,
            amount_cents INTEGER NOT NULL,
            currency TEXT NOT NULL DEFAULT 'USD',
            status TEXT NOT NULL,
            provider_ref TEXT,
            created_at TIMESTAMP DEFAULT NOW()
        )
    """))

This table gives you traceability. If an agent retries a charge, you can check whether the previous attempt already succeeded.

3. •Define a payment tool that the agent can call.

LangChain agents work well when external operations are exposed as tools. In production, this tool should call your payment provider SDK or API, then return structured output.

from langchain_core.tools import tool
import uuid

@tool
def charge_payment(user_id: str, amount_cents: int, currency: str = "USD") -> dict:
    """
    Charge a customer and return a structured result.
    Replace the stubbed implementation with your payment provider SDK.
    """
    provider_ref = f"pay_{uuid.uuid4().hex[:12]}"
    status = "succeeded"

    with engine.begin() as conn:
        conn.execute(
            text("""
                INSERT INTO payment_events (user_id, action, amount_cents, currency, status, provider_ref)
                VALUES (:user_id, :action, :amount_cents, :currency, :status, :provider_ref)
            """),
            {
                "user_id": user_id,
                "action": "charge",
                "amount_cents": amount_cents,
                "currency": currency,
                "status": status,
                "provider_ref": provider_ref,
            },
        )

    return {
        "user_id": user_id,
        "amount_cents": amount_cents,
        "currency": currency,
        "status": status,
        "provider_ref": provider_ref,
    }

If you already have a LangChain payments integration package from your provider, swap the stubbed logic for that SDK’s charge method and keep the same return shape.

4. •Wire the tool into a LangChain agent.

Use ChatOpenAI plus create_tool_calling_agent and AgentExecutor. That gives the model permission to call your payment tool when needed.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import create_tool_calling_agent, AgentExecutor

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a payments assistant. Only charge when the request is explicit."),
    ("human", "{input}"),
])

tools = [charge_payment]
agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = executor.invoke({
    "input": "Charge user_123 $19.99 for subscription renewal."
})

print(result["output"])

In production, add guardrails before any tool call:

• •validate amount limits
• •require explicit user confirmation
• •store an idempotency key in PostgreSQL
• •log every refusal and retry

5. •Query PostgreSQL to build agent memory or reconciliation logic.

Once charges are stored in Postgres, the agent can answer questions like “did this customer already pay?” or “what failed last night?”

def get_recent_payments(user_id: str):
    with engine.begin() as conn:
        rows = conn.execute(
            text("""
                SELECT action, amount_cents, currency, status, provider_ref, created_at
                FROM payment_events
                WHERE user_id = :user_id
                ORDER BY created_at DESC
                LIMIT 10
            """),
            {"user_id": user_id},
        ).fetchall()

    return [dict(row._mapping) for row in rows]

This is where PostgreSQL earns its place. The model reasons; Postgres remembers.

Testing the Integration

Run one end-to-end test: invoke the agent with a charge request and verify that Postgres recorded it.

test_result = executor.invoke({
    "input": "Charge user_999 $5.00 for one-time verification."
})

print(test_result["output"])
print(get_recent_payments("user_999"))

Expected output:

Payment completed for user_999.
[{'action': 'charge', 'amount_cents': 500, 'currency': 'USD', 'status': 'succeeded', 'provider_ref': 'pay_a1b2c3d4e5f6', 'created_at': datetime.datetime(...)}]

If you get an empty list:

• •confirm DATABASE_URL
• •confirm the insert ran inside engine.begin()
• •check that the agent actually called charge_payment
• •inspect verbose=True logs from AgentExecutor

Real-World Use Cases

• •

  Subscription billing assistants

  • •An agent can explain invoice status to users and trigger charges only after policy checks pass.
  • •PostgreSQL stores billing history and retry metadata for compliance.
• •

  Refund triage bots

  • •The agent inspects order context from Postgres before deciding whether a refund is allowed.
  • •It then calls the payments tool and writes an auditable refund event.
• •

  Collections and dunning workflows

  • •The agent checks failed-payment history in Postgres.
  • •It generates next-step actions like retry charge, notify customer, or escalate to support.

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.