Payment Processing Tools

Payments are the lifeblood of e-commerce. When your AI agent can process payments, issue refunds, and manage subscriptions, it becomes a powerful tool for customer service and automation.

In this lesson, we'll build MCP tools that integrate with Stripe - the most widely-used payment platform, which has official MCP server support.

Payment Tool Architecture

┌──────────────────────────────────────────────────────────────────┐
│                     AI AGENT CONVERSATION                        │
│  "Refund the customer $50 for order #12345"                     │
└─────────────────────────────────┬────────────────────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────┐
│                    PAYMENT MCP TOOLS                             │
├───────────────┬───────────────┬───────────────┬─────────────────┤
│ create_charge │ process_refund│ get_balance   │ list_charges    │
│ get_payment   │ create_payout │ get_customer  │ update_customer │
└───────────────┴───────────────┴───────────────┴─────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────┐
│                      STRIPE API                                  │
│           Payment Intents • Charges • Refunds • Customers        │
└──────────────────────────────────────────────────────────────────┘

Core Payment Tools

1. Create Payment Intent

Payment Intents are Stripe's recommended way to collect payments. They handle the full payment lifecycle including 3D Secure authentication.

from mcp.server import Server
from mcp import types
import stripe
import json

app = Server("payment-tools")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="create_payment_intent",
            description="""Creates a PaymentIntent to collect payment from a customer.
            Use when: Customer is ready to pay for their cart/order.
            Returns: PaymentIntent with client_secret for frontend confirmation.
            Note: Amount is in cents (e.g., 2999 = $29.99).""",
            inputSchema={
                "type": "object",
                "properties": {
                    "amount": {
                        "type": "integer",
                        "description": "Amount in cents (e.g., 2999 for $29.99)",
                        "minimum": 50
                    },
                    "currency": {
                        "type": "string",
                        "enum": ["usd", "eur", "gbp", "cad", "aud"],
                        "default": "usd",
                        "description": "Three-letter ISO currency code"
                    },
                    "customer_id": {
                        "type": "string",
                        "description": "Stripe customer ID (e.g., 'cus_abc123')"
                    },
                    "payment_method_id": {
                        "type": "string",
                        "description": "Saved payment method to use. Omit for new card."
                    },
                    "description": {
                        "type": "string",
                        "description": "Description shown on customer's statement"
                    },
                    "metadata": {
                        "type": "object",
                        "description": "Key-value pairs (e.g., order_id, product_ids)",
                        "additionalProperties": {"type": "string"}
                    },
                    "idempotency_key": {
                        "type": "string",
                        "description": "Unique key to prevent duplicate charges. Required."
                    }
                },
                "required": ["amount", "customer_id", "idempotency_key"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "create_payment_intent":
        try:
            intent = stripe.PaymentIntent.create(
                amount=arguments["amount"],
                currency=arguments.get("currency", "usd"),
                customer=arguments["customer_id"],
                payment_method=arguments.get("payment_method_id"),
                description=arguments.get("description"),
                metadata=arguments.get("metadata", {}),
                idempotency_key=arguments["idempotency_key"],
                automatic_payment_methods={"enabled": True}
            )

            return [types.TextContent(
                type="text",
                text=json.dumps({
                    "success": True,
                    "payment_intent_id": intent.id,
                    "client_secret": intent.client_secret,
                    "status": intent.status,
                    "amount": intent.amount,
                    "currency": intent.currency
                })
            )]
        except stripe.error.StripeError as e:
            return [types.TextContent(
                type="text",
                text=json.dumps({
                    "success": False,
                    "error": str(e),
                    "error_code": e.code if hasattr(e, 'code') else None
                })
            )]

2. Process Refund Tool

types.Tool(
    name="process_refund",
    description="""Creates a refund for a previous charge or payment intent.
    Use when: Customer requests a refund, order is cancelled, or item is returned.
    Returns: Refund details including status and amount.
    Note: Partial refunds are supported - specify amount less than original charge.""",
    inputSchema={
        "type": "object",
        "properties": {
            "payment_intent_id": {
                "type": "string",
                "description": "The PaymentIntent ID to refund (e.g., 'pi_abc123')"
            },
            "charge_id": {
                "type": "string",
                "description": "Alternative: The Charge ID to refund (e.g., 'ch_abc123')"
            },
            "amount": {
                "type": "integer",
                "description": "Amount to refund in cents. Omit for full refund.",
                "minimum": 1
            },
            "reason": {
                "type": "string",
                "enum": ["duplicate", "fraudulent", "requested_by_customer"],
                "description": "Reason for the refund"
            },
            "metadata": {
                "type": "object",
                "description": "Additional data (e.g., refund_ticket_id, agent_id)",
                "additionalProperties": {"type": "string"}
            },
            "idempotency_key": {
                "type": "string",
                "description": "Unique key to prevent duplicate refunds. Required."
            }
        },
        "required": ["idempotency_key"],
        "oneOf": [
            {"required": ["payment_intent_id"]},
            {"required": ["charge_id"]}
        ]
    }
)

elif name == "process_refund":
    try:
        refund_params = {
            "reason": arguments.get("reason"),
            "metadata": arguments.get("metadata", {}),
            "idempotency_key": arguments["idempotency_key"]
        }

        if "payment_intent_id" in arguments:
            refund_params["payment_intent"] = arguments["payment_intent_id"]
        elif "charge_id" in arguments:
            refund_params["charge"] = arguments["charge_id"]

        if "amount" in arguments:
            refund_params["amount"] = arguments["amount"]

        refund = stripe.Refund.create(**refund_params)

        return [types.TextContent(
            type="text",
            text=json.dumps({
                "success": True,
                "refund_id": refund.id,
                "amount": refund.amount,
                "currency": refund.currency,
                "status": refund.status,
                "reason": refund.reason
            })
        )]
    except stripe.error.StripeError as e:
        return [types.TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": str(e),
                "error_code": getattr(e, 'code', None)
            })
        )]

3. Get Customer Payment Methods

types.Tool(
    name="list_payment_methods",
    description="""Lists saved payment methods for a customer.
    Use when: Customer wants to pay with a saved card, or to show payment options.
    Returns: Array of payment methods with last 4 digits, brand, and expiry.""",
    inputSchema={
        "type": "object",
        "properties": {
            "customer_id": {
                "type": "string",
                "description": "Stripe customer ID"
            },
            "type": {
                "type": "string",
                "enum": ["card", "bank_account", "us_bank_account"],
                "default": "card",
                "description": "Type of payment methods to list"
            }
        },
        "required": ["customer_id"]
    }
)

elif name == "list_payment_methods":
    payment_methods = stripe.PaymentMethod.list(
        customer=arguments["customer_id"],
        type=arguments.get("type", "card")
    )

    methods = []
    for pm in payment_methods.data:
        if pm.type == "card":
            methods.append({
                "id": pm.id,
                "type": "card",
                "brand": pm.card.brand,
                "last4": pm.card.last4,
                "exp_month": pm.card.exp_month,
                "exp_year": pm.card.exp_year,
                "is_default": pm.id == pm.customer  # simplified
            })

    return [types.TextContent(
        type="text",
        text=json.dumps({"payment_methods": methods})
    )]

TypeScript Implementation

Here's the same payment tools in TypeScript:

import Stripe from "stripe";
import { Server } from "@modelcontextprotocol/sdk/server";

const stripe = new Stripe(process.env.STRIPE_SECRET_KEY!);
const server = new Server({ name: "payment-tools", version: "1.0.0" });

const paymentTools = [
  {
    name: "create_payment_intent",
    description: "Creates a PaymentIntent to collect payment from a customer...",
    inputSchema: {
      type: "object" as const,
      properties: {
        amount: { type: "integer", minimum: 50, description: "Amount in cents" },
        currency: { type: "string", enum: ["usd", "eur", "gbp"], default: "usd" },
        customer_id: { type: "string", description: "Stripe customer ID" },
        idempotency_key: { type: "string", description: "Unique key for idempotency" }
      },
      required: ["amount", "customer_id", "idempotency_key"]
    }
  },
  {
    name: "process_refund",
    description: "Creates a refund for a previous charge...",
    inputSchema: {
      type: "object" as const,
      properties: {
        payment_intent_id: { type: "string" },
        amount: { type: "integer", minimum: 1 },
        reason: { type: "string", enum: ["duplicate", "fraudulent", "requested_by_customer"] },
        idempotency_key: { type: "string" }
      },
      required: ["payment_intent_id", "idempotency_key"]
    }
  },
  {
    name: "get_account_balance",
    description: "Retrieves current Stripe account balance...",
    inputSchema: {
      type: "object" as const,
      properties: {},
      required: []
    }
  }
];

server.setRequestHandler("tools/list", async () => ({ tools: paymentTools }));

server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  try {
    switch (name) {
      case "create_payment_intent": {
        const intent = await stripe.paymentIntents.create({
          amount: args.amount,
          currency: args.currency || "usd",
          customer: args.customer_id,
          automatic_payment_methods: { enabled: true }
        }, {
          idempotencyKey: args.idempotency_key
        });

        return {
          content: [{
            type: "text",
            text: JSON.stringify({
              success: true,
              payment_intent_id: intent.id,
              client_secret: intent.client_secret,
              status: intent.status,
              amount: intent.amount
            })
          }]
        };
      }

      case "process_refund": {
        const refund = await stripe.refunds.create({
          payment_intent: args.payment_intent_id,
          amount: args.amount,
          reason: args.reason
        }, {
          idempotencyKey: args.idempotency_key
        });

        return {
          content: [{
            type: "text",
            text: JSON.stringify({
              success: true,
              refund_id: refund.id,
              amount: refund.amount,
              status: refund.status
            })
          }]
        };
      }

      case "get_account_balance": {
        const balance = await stripe.balance.retrieve();

        return {
          content: [{
            type: "text",
            text: JSON.stringify({
              available: balance.available,
              pending: balance.pending
            })
          }]
        };
      }

      default:
        throw new Error(`Unknown tool: ${name}`);
    }
  } catch (error) {
    const stripeError = error as Stripe.errors.StripeError;
    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          success: false,
          error: stripeError.message,
          code: stripeError.code
        })
      }]
    };
  }
});

Using Stripe's Official MCP Server

Stripe provides an official MCP server you can use directly:

# Install and run locally
npx -y @stripe/mcp --api-key=sk_test_...

# Or connect to Stripe's hosted MCP server
# https://mcp.stripe.com (uses OAuth)

Available Tools in Stripe MCP

Security Best Practices

1. Always Use Idempotency Keys

# REQUIRED for any mutating operation
idempotency_key = f"order_{order_id}_{uuid.uuid4()}"

stripe.PaymentIntent.create(
    amount=2999,
    currency="usd",
    customer="cus_abc123",
    idempotency_key=idempotency_key  # Prevents duplicate charges
)

2. Never Log Sensitive Data

# BAD - logs full card number
logger.info(f"Processing payment for card {card_number}")

# GOOD - logs only last 4 digits
logger.info(f"Processing payment for card ending in {last_four}")

3. Use Webhook Verification

# Verify webhook signatures to prevent spoofing
@app.post("/stripe/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, webhook_secret
        )
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    # Process verified event
    handle_event(event)

4. Implement Amount Limits

MAX_SINGLE_CHARGE = 100000  # $1,000 in cents
MAX_DAILY_REFUND = 500000   # $5,000 in cents

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "create_payment_intent":
        if arguments["amount"] > MAX_SINGLE_CHARGE:
            return error_response(
                "Amount exceeds maximum allowed. Please contact support."
            )

Real-World Conversation Flow

Here's how an AI agent uses payment tools in a real conversation:

Customer: "I want to buy the blue sneakers, size 10"

AI Agent: [Thinking]
  1. Look up product → get_product(sku="SNKR-BLUE-10")
  2. Check inventory → check_inventory(product_id="SNKR-BLUE-10")
  3. Get customer → get_customer(email="customer@email.com")

AI Agent: "Great choice! The Blue Running Sneakers (Size 10) are $89.99
and in stock. I found your saved Visa ending in 4242.
Would you like me to complete the purchase?"

Customer: "Yes, use that card"

AI Agent: [Calls tool]
  create_payment_intent(
    amount=8999,
    customer_id="cus_abc123",
    payment_method_id="pm_xyz789",
    idempotency_key="order_12345_abc",
    metadata={"order_id": "12345", "product": "SNKR-BLUE-10"}
  )

AI Agent: "Perfect! Your order is confirmed. I've charged $89.99 to your
Visa ending in 4242. Order #12345 will ship within 2 business days.
You'll receive a confirmation email shortly."

Key Takeaways

Next up: Inventory Management Tools - Building tools to check stock levels, reserve inventory, and manage warehouse operations.