LangChain Stripe Integration: Safe Agent Payments with Policy Enforcement

Why the default LangChain + Stripe setup is dangerous

LangChain makes it straightforward to wrap any Python function as a Tool and drop it into an agent. The Stripe Python SDK is equally easy to call. Put the two together and you have an agent that can create charges, issue refunds, and manage subscriptions on command.

The standard setup looks like this:

from langchain_core.tools import tool
import stripe

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # full secret key

@tool
def charge_customer(customer_id: str, amount_cents: int, description: str) -> str:
    """Create a Stripe charge for the given customer."""
    charge = stripe.Charge.create(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=description,
    )
    return f"Charge created: {charge.id}"

This works in a demo. In production it has three problems:

• No spend cap. If your agent enters a retry loop — a common failure mode when the LLM decides a sub-task "might not have worked" — it will issue real charges until you notice or your Stripe account hits its fraud threshold.
• No endpoint allowlist. The same tool can be repurposed (by a sufficiently creative LLM) to call stripe.PaymentMethod.detach() or stripe.Customer.delete() if those operations are reachable through related objects.
• No audit log. Stripe has transaction history, but it does not tell you which agent run caused each charge, what the agent's reasoning was, or whether the charge was within the parameters you intended.

Step 1 — Scope the Stripe API key

The first safeguard costs nothing and takes two minutes. Instead of passing your full Stripe secret key to the agent, create a Stripe restricted API key scoped to exactly the operations your agent needs.

For a billing agent that only creates charges against existing customers:

# In Stripe Dashboard → Developers → API keys → Create restricted key
#
# Permissions needed for a charge-only billing agent:
#   Charges           → Write
#   Customers         → Read        (lookup only, no delete)
#   Payment intents   → Write       (if using newer payment flow)
#   Payment methods   → Read        (attach, not detach)
#
# Everything else → None
#
# Name: "billing-agent-production" or "billing-agent-[run-id]" per run

A restricted key cannot call stripe.Customer.delete(), cannot issue refunds unless you grant Refunds → Write, and cannot list or modify other resources. The blast radius of a stuck agent shrinks to the specific subset of Stripe you explicitly permitted.

See the full Stripe restricted key permissions reference for the complete list of toggles and which minimum set to use per agent archetype.

Step 2 — Add a spend cap to the tool

A scoped key limits what the agent can do; a spend cap limits how much. You can implement a simple in-process cap using a thread-local accumulator:

import threading
from langchain_core.tools import tool
import stripe

stripe.api_key = os.environ["STRIPE_RESTRICTED_KEY"]

_spend_lock = threading.Lock()
_session_spend_cents = 0
SESSION_CAP_CENTS = 10_000  # $100 per agent session

@tool
def charge_customer(customer_id: str, amount_cents: int, description: str) -> str:
    """Create a Stripe charge. Enforces a $100 per-session spend cap."""
    global _session_spend_cents

    if amount_cents <= 0:
        raise ValueError("amount_cents must be positive")

    with _spend_lock:
        projected = _session_spend_cents + amount_cents
        if projected > SESSION_CAP_CENTS:
            raise RuntimeError(
                f"Spend cap exceeded: would reach ${projected/100:.2f}, "
                f"cap is ${SESSION_CAP_CENTS/100:.2f}"
            )

    charge = stripe.Charge.create(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=description,
        idempotency_key=f"agent-{os.environ.get('RUN_ID','default')}-{customer_id}-{amount_cents}",
    )

    with _spend_lock:
        _session_spend_cents += amount_cents

    return f"Charge created: {charge.id} (session total: ${_session_spend_cents/100:.2f})"

Note the idempotency key using a run-scoped prefix. See the Stripe idempotency key guide for agents for why this matters when agents retry.

The in-process cap works for single-instance agents. It fails for:

• Multi-instance deployments (two processes share no state)
• Agent restarts mid-session (cap resets to zero)
• CrewAI or AutoGen multi-agent crews where sub-agents run in parallel

For those cases you need a cap that lives outside the process — which is what a proxy layer provides.

Step 3 — Route through a governed proxy

A proxy between the agent and Stripe enforces policies in a single place that survives restarts, scales across instances, and logs every call to a store the agent cannot write to.

With Keybrake, you issue the agent a vault_key instead of a real Stripe secret, attach a policy, and point the agent's Stripe base URL at the proxy:

import stripe

# Point the Stripe SDK at the Keybrake proxy instead of api.stripe.com
stripe.api_key = os.environ["VAULT_KEY"]         # vault_key_xxx, not sk_live_xxx
stripe.api_base = "https://proxy.keybrake.com/stripe"

# Everything else is unchanged — same SDK calls, same response shapes
charge = stripe.Charge.create(
    amount=5000,
    currency="usd",
    customer="cus_abc123",
    description="Monthly billing",
)

The vault key policy (set once via the Keybrake API or dashboard):

{
  "vendor": "stripe",
  "daily_usd_cap": 500,
  "allowed_endpoints": [
    "POST /v1/charges",
    "GET /v1/charges/*",
    "GET /v1/customers/*"
  ],
  "merchant_allowlist": ["acct_abc123"],
  "expires_at": "2026-07-01T00:00:00Z"
}

Any call outside allowed_endpoints returns 403 before it reaches Stripe. Any call that would push the daily total past daily_usd_cap is blocked and logged. Revoke the vault key mid-run with one API call — no Stripe key rotation needed.

Full LangChain tool with proxy routing

Putting it together: a StripeChargeTool that uses the Keybrake proxy and includes input validation, idempotency, and error propagation back to the LLM:

import os
import stripe
from langchain_core.tools import tool
from pydantic import BaseModel, Field

# Route all Stripe calls through the proxy
stripe.api_key = os.environ["VAULT_KEY"]
stripe.api_base = "https://proxy.keybrake.com/stripe"

RUN_ID = os.environ.get("AGENT_RUN_ID", "dev")


class ChargeInput(BaseModel):
    customer_id: str = Field(description="Stripe customer ID (cus_xxx)")
    amount_cents: int = Field(ge=50, le=99900, description="Amount in cents (50–99900)")
    description: str = Field(max_length=200, description="Human-readable charge description")


@tool(args_schema=ChargeInput)
def create_stripe_charge(customer_id: str, amount_cents: int, description: str) -> str:
    """
    Create a Stripe charge for a customer. Returns the charge ID on success.
    The charge is subject to the vault key's daily spend cap and endpoint allowlist.
    """
    try:
        charge = stripe.Charge.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=description,
            idempotency_key=f"{RUN_ID}-{customer_id}-{amount_cents}",
        )
        return f"Charge {charge.id} created for ${amount_cents/100:.2f}"

    except stripe.error.AuthenticationError:
        return "Error: vault key rejected — check VAULT_KEY env var"
    except stripe.error.PermissionError as e:
        return f"Error: operation not permitted by policy — {e.user_message}"
    except stripe.error.CardError as e:
        return f"Card declined: {e.user_message}"
    except stripe.error.StripeError as e:
        return f"Stripe error: {e.user_message}"

Wire it into a LangChain agent:

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

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [create_stripe_charge]

prompt = ChatPromptTemplate.from_messages([
    ("system", (
        "You are a billing agent. You can create Stripe charges for customers. "
        "Always confirm the customer ID and amount before charging. "
        "Never charge more than $100 without explicit user confirmation."
    )),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

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

Testing the governed tool

Two tests that verify policy enforcement without hitting real Stripe:

import pytest
from unittest.mock import patch, MagicMock

def test_charge_within_policy():
    """Successful charge returns charge ID."""
    mock_charge = MagicMock()
    mock_charge.id = "ch_test123"

    with patch("stripe.Charge.create", return_value=mock_charge):
        result = create_stripe_charge.invoke({
            "customer_id": "cus_abc",
            "amount_cents": 5000,
            "description": "Monthly plan",
        })
    assert "ch_test123" in result

def test_proxy_blocks_unpermitted_endpoint():
    """Proxy 403 on disallowed endpoint propagates as a PermissionError message."""
    import stripe as stripe_mod
    perm_err = stripe_mod.error.PermissionError(
        message="Endpoint not in allowlist",
        http_status=403,
        json_body={"error": {"message": "Endpoint not in allowlist"}},
    )
    with patch("stripe.Charge.create", side_effect=perm_err):
        result = create_stripe_charge.invoke({
            "customer_id": "cus_abc",
            "amount_cents": 5000,
            "description": "Should be blocked",
        })
    assert "not permitted by policy" in result

def test_pydantic_blocks_zero_amount():
    """Input validation rejects zero-cent charges before they reach Stripe."""
    with pytest.raises(Exception):
        create_stripe_charge.invoke({
            "customer_id": "cus_abc",
            "amount_cents": 0,
            "description": "Zero charge attempt",
        })

Comparison: bare key vs restricted key vs vault key

The restricted key and vault key are complementary, not competing. Use both: the restricted key limits the surface at the Stripe level, the vault key adds spend caps, per-run revoke, and an audit trail that the agent cannot modify.

Frequently asked questions

Does changing stripe.api_base break the SDK?

No. The Stripe Python SDK uses api_base as the root URL for all API requests. The proxy forwards the request to api.stripe.com after policy enforcement, and returns the unmodified Stripe response. The SDK never knows it is talking to a proxy.

What happens to the Stripe SDK's retry logic?

The SDK retries on network errors and 5xx responses. The proxy only blocks (4xx) on policy violations — so retries on genuine network errors still reach Stripe. Policy-blocked calls return 403, which the SDK does not retry.

Can I use this with LangChain's async streaming?

Yes. The tool is a standard synchronous Python function, which LangChain runs in a thread pool when invoked from an async context via ainvoke. The Stripe SDK call inside the tool is also synchronous — for async-native Stripe calls, use stripe.Charge.create_async() (available in stripe-python >= 7.0) inside an async def tool decorated with @tool.

How do I issue a different vault key per agent run?

Call the Keybrake API before each run to mint a vault key scoped to that run's parameters. Store the returned vault_key_xxx in the agent's environment. The key expires automatically at expires_at — no cleanup code needed.

What is the latency overhead of the proxy?

Policy lookup and spend accounting add ~2–5 ms per request (SQLite read + write on the same host). Stripe API calls typically take 150–400 ms, so the overhead is below measurement noise for production workloads.

Does this work with LangGraph and multi-agent crews?

Yes. Each node or sub-agent in a LangGraph workflow or CrewAI crew that needs Stripe access gets its own vault key with its own spend cap. The proxy aggregates spend across all keys so you can also set a project-level cap that applies across the entire workflow.

Wrapping up

A LangChain Stripe integration is a few lines of Python. A safe LangChain Stripe integration adds a scoped key (cuts endpoint surface), an idempotency key (prevents double-charges on retry), and a proxy layer (enforces spend caps and produces an audit trail that survives agent restarts and multi-instance deployments).

The proxy is the piece that is hardest to build yourself and easiest to overlook — until an agent bills a customer twice or a runaway loop burns through your Stripe balance at 3 a.m.

If you are building a LangChain agent that touches Stripe, try the Keybrake proxy — point stripe.api_base at it, get a vault key, and you have a spend cap and audit log in place before your first production run.