Automating Dispute Management with Agents SDK and Stripe API (part 4)

In the "Best practices for defining functions" section in Function calling guide, we can read the following:

• Offload the burden from the model and use code where possible.

  • Don't make the model fill arguments you already know.

  • Combine functions that are always called in sequence.

Let's try to follow that.

I modified the retrieve_payment_intent function tool so it no longer takes the payment intent ID as an argument. We already have the ID in process_dispute, so there's no reason for the LLM to discover it.

To get this working, I defined it locally inside process_dispute body along with the Triage Agent, which is now also local. On the first turn, the LLM calls retrieve_payment_intent with no argument.

The rest of the code remains the same.

async def process_dispute(payment_intent_id):
    """Retrieve and process dispute data for a given PaymentIntent."""
    disputes_list = stripe.Dispute.list(payment_intent=payment_intent_id)
    if not disputes_list.data:
        logger.warning("No dispute data found for PaymentIntent: %s", payment_intent_id)
        return None

    dispute_data = disputes_list.data[0]

    relevant_data = {
        "dispute_id": dispute_data.get("id"),
        "amount": dispute_data.get("amount"),
        "due_by": dispute_data.get("evidence_details", {}).get("due_by"),
        "payment_intent": dispute_data.get("payment_intent"),
        "reason": dispute_data.get("reason"),
        "status": dispute_data.get("status"),
        "card_brand": dispute_data.get("payment_method_details", {}).get("card", {}).get("brand")
    }

    event_str = json.dumps(relevant_data)
    @function_tool
    async def retrieve_payment_intent() -> dict:
        """
        Retrieve a Stripe payment intent by ID.
        Returns the payment intent object on success or an empty dictionary on failure.
        """
        try:
            return stripe.PaymentIntent.retrieve(payment_intent_id)
        except stripe.error.StripeError as e:
            logger.error(f"Stripe error occurred while retrieving payment intent: {e}")
            return {}

    triage_agent = Agent(
        name="Triage Agent",
        instructions=(
            "Please do the following:\n"
            "1. Find the order ID from the payment intent's metadata.\n"
            "2. Retrieve detailed information about the order (e.g., shipping status).\n"
            "3. If the order has shipped, escalate this dispute to the investigator agent.\n"
            "4. If the order has not shipped, accept the dispute.\n"
        ),
        model="gpt-4.1",
        tools=[retrieve_payment_intent, get_order],
        handoffs=[accept_dispute_agent, investigator_agent],
    )


    result = await Runner.run(triage_agent, input=event_str)
    logger.info("WORKFLOW RESULT: %s", result.final_output)

    return relevant_data, result.final_output

payment = stripe.PaymentIntent.create(
  amount=2000,
  currency="usd",
  payment_method = "pm_card_createDisputeProductNotReceived",
  confirm=True,
  metadata={"order_id": "1234"},
  off_session=True,
  automatic_payment_methods={"enabled": True},
)

asyncio.run(process_dispute(payment.id))

Dispute Details:
- Dispute ID: dp_1RoLJ5QuMzbiv2kfKKLwQcgv
- Amount: $20.00 (2000 cents)
- Reason for Dispute: Product not received

Order Details:
- Fulfillment status of the order: Not shipped

Reasoning for closing the dispute:
The dispute is being accepted because the customer claims they did not
receive the product, and the order records confirm that the item has
not been shipped. Since the merchandise was not fulfilled, we are
accepting the dispute as valid.

payment = stripe.PaymentIntent.create(
  amount=2000,
  currency="usd",
  payment_method = "pm_card_createDispute",
  confirm=True,
  metadata={"order_id": "1121"},
  off_session=True,
  automatic_payment_methods={"enabled": True},
)

asyncio.run(process_dispute(payment.id))

Dispute Details:
- Dispute ID: dp_1RoLKZQuMzbiv2kf5bfNx6G7
- Amount: $20.00 (2000 cents)
- Reason for Dispute: Fraudulent
- Card Brand: Visa

Payment & Order Details:
- Fulfillment Status of the Order: Delivered
- Shipping Carrier and Tracking Number: UPS, Tracking #1Z999AA10123456784
- Confirmation of TOS Acceptance: Accepted on 2023-01-01, IP: 192.168.1.1

Email and Phone Records:

Relevant Email Threads:
- Subject: Order #1121
  - Body: "Hey, I know you don't accept refunds but the sneakers don't
  fit and I'd like a refund"

Relevant Phone Logs:
1. 2023-02-28 10:10 (7 min): Requested refund for order #1121, I told
him we were unable to refund the order because it was final sale.
2. 2023-03-14 15:24 (5 min): Asked about status of order #1121.

Summary of Findings:

Although the dispute reason is listed as "fraudulent," the email and
phone records indicate the customer received the product and was
unsatisfied with the fit, requesting a refund. The product was
delivered, and the customer accepted the terms of sale (final sale/no
refunds). No communications indicate fraudulent activity.

Here, I removed the Triage Agent altogether:

triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "Please do the following:\n"
        "1. Find the order ID from the payment intent's metadata.\n"
        "2. Retrieve detailed information about the order (e.g., shipping status).\n"
        "3. If the order has shipped, escalate this dispute to the investigator agent.\n"
        "4. If the order has not shipped, accept the dispute.\n"
    ),
    model="gpt-4o",
    tools=[retrieve_payment_intent, get_order],
    handoffs=[accept_dispute_agent, investigator_agent],
)

To get the fulfillment_details information, I directly called retrieve_payment_intent and get_order in process_dispute; now as regular Python functions, not function tools.

If fulfillment_details was not_shipped, I ran a workflow starting with the "Accept Dispute Agent". If not, it began with the "Dispute Intake Agent".

This version works the same.

However, if the value of fulfillment_details in the order could be unshipped or not_dispatched for instance instead of not_shipped, we would have to update our condition condition logic. If we let the LLM decide, this wouldn't be necessary.

I've also been careful to pass both order and payment_intent data to the LLM, since in the original code a Triage Agent handoff shares the full context (including function calls and their inputs/outputs) by default. You can change this with input filters (see handoffs docs).

def get_order(order_id: int) -> str:
    """
    Retrieve an order by ID from a predefined list of orders.
    Returns the corresponding order object or 'No order found'.
    """
    # ...

async def retrieve_payment_intent(payment_intent_id):
    """
    Retrieve a Stripe payment intent by ID.
    Returns the payment intent object on success or an empty dictionary on failure.
    """
    # ...

async def process_dispute(payment_intent_id):
    """Retrieve and process dispute data for a given PaymentIntent."""
    disputes_list = stripe.Dispute.list(payment_intent=payment_intent_id)
    if not disputes_list.data:
        logger.warning("No dispute data found for PaymentIntent: %s", payment_intent_id)
        return None
    dispute_data = disputes_list.data[0]

    payment_intent = await retrieve_payment_intent(payment_intent_id)
    order_id = int(payment_intent.get("metadata").get("order_id"))
    order = get_order(order_id)
    relevant_data = {
        "dispute_id": dispute_data.get("id"),
        "amount": dispute_data.get("amount"),
        "due_by": dispute_data.get("evidence_details", {}).get("due_by"),
        "reason": dispute_data.get("reason"),
        "status": dispute_data.get("status"),
        "card_brand": dispute_data.get("payment_method_details", {}).get("card", {}).get("brand"),
        # ---
        "payment_intent": payment_intent.to_dict_recursive(),
        "order": order
    }
    event_str = json.dumps(relevant_data)

    if order.get("fulfillment_details") == "not_shipped":
        result = await Runner.run(accept_dispute_agent, input=event_str)
    else:
        result = await Runner.run(investigator_agent, input=event_str)

    logger.info("WORKFLOW RESULT: %s", result.final_output)

    return relevant_data, result.final_output

Note that to_dict_recursive() will be deprecated:

DeprecationWarning: For internal stripe-python use only. The public interface will be removed in a future version.

Finally, I removed the "Accept Dispute Agent":

accept_dispute_agent = Agent(
    name="Accept Dispute Agent",
    instructions=(
        "You are an agent responsible for accepting disputes. Please do the following:\n"
        "1. Use the provided dispute ID to close the dispute.\n"
        "2. Provide a short explanation of why the dispute is being accepted.\n"
        "3. Reference any relevant order details (e.g., unfulfilled order, etc.) retrieved from the database.\n\n"
        "Then, produce your final output in this exact format:\n\n"
        "Dispute Details:\n"
        "- Dispute ID\n"
        "- Amount\n"
        "- Reason for Dispute\n\n"
        "Order Details:\n"
        "- Fulfillment status of the order\n\n"
        "Reasoning for closing the dispute\n"
    ),
    model="gpt-4.1",
    tools=[close_dispute]
)

To get the final explanation of the reports, I called the Responses API using openai packages.

I could have done the same with "Dispute Intake Agent", but I left it unchanged:

investigator_agent = Agent(
    name="Dispute Intake Agent",
    instructions=(
        "As a dispute investigator, please compile the following details in your final output:\n\n"
        "Dispute Details:\n"
        "- Dispute ID\n"
        "- Amount\n"
        "- Reason for Dispute\n"
        "- Card Brand\n\n"
        "Payment & Order Details:\n"
        "- Fulfillment status of the order\n"
        "- Shipping carrier and tracking number\n"
        "- Confirmation of TOS acceptance\n\n"
        "Email and Phone Records:\n"
        "- Any relevant email threads (include the full body text)\n"
        "- Any relevant phone logs\n"
    ),
    model="gpt-4.1",
    tools=[get_emails, get_phone_logs]
)

import asyncio
import os
import logging
import json
from dotenv import load_dotenv
from agents import Agent, Runner, function_tool  # Only import what you need
import stripe
from typing_extensions import TypedDict, Any
from openai import OpenAI
# Load environment variables from .env file
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Set Stripe API key from environment variables
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

@function_tool
def get_phone_logs(phone_number: str) -> list:
    """
    Return a list of phone call records for the given phone number.
    Each record might include call timestamps, durations, notes,
    and an associated order_id if applicable.
    """
    # ...

@function_tool
def get_emails(email: str) -> list:
    """
    Return a list of email records for the given email address.
    """
    # ...


investigator_agent = Agent(
    name="Dispute Intake Agent",
    instructions=(...),
    model="gpt-4.1",
    tools=[get_emails, get_phone_logs]
)

async def close_dispute(dispute_id: str) -> dict:
    """
    Close a Stripe dispute by ID.
    Returns the dispute object on success or an empty dictionary on failure.
    """
    try:
        return stripe.Dispute.close(dispute_id)
    except stripe.error.StripeError as e:
        logger.error(f"Stripe error occurred while closing dispute: {e}")
        return {}

async def accept_dispute(data):
    dispute_id = data.get("dispute_id")
    await close_dispute(dispute_id)

    amount = data.get("amount")
    reason = data.get("reason").replace("_", " ")
    fulfillment_details = data.get("order").get("fulfillment_details").replace("_", " ")
    report = (
        "Dispute Details:\n"
        f"- Dispute ID: {dispute_id}\n"
        f"- Amount: {amount}\n"
        f"- Reason for Dispute: {reason}\n\n"
        "Order Details:\n"
        f"- Fulfillment status of the order: {fulfillment_details}\n\n")

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    response = client.responses.create(
        model="gpt-4.1",
        input=[{"role": "user", "content": report}],
        instructions="Give a short explanation for closing the dispute.  No title, just the paragraph."
    )
    reasoning_for_closing_the_dispute = response.output_text
    return f"{report}Reasoning for closing the dispute:\n{reasoning_for_closing_the_dispute}"


def get_order(order_id: int) -> str:
    """
    Retrieve an order by ID from a predefined list of orders.
    Returns the corresponding order object or 'No order found'.
    """
    orders = [
        {
            "order_id": 1234,
            "fulfillment_details": "not_shipped"
        },
        {
            "order_id": 9101,
            "fulfillment_details": "shipped",
            "tracking_info": {
                "carrier": "FedEx",
                "tracking_number": "123456789012"
            },
            "delivery_status": "out for delivery"
        },
        {
            "order_id": 1121,
            "fulfillment_details": "delivered",
            "customer_id": "cus_PZ1234567890",
            "customer_phone": "+15551234567",
            "order_date": "2023-01-01",
            "customer_email": "customer1@example.com",
            "tracking_info": {
                "carrier": "UPS",
                "tracking_number": "1Z999AA10123456784",
                "delivery_status": "delivered"
            },
            "shipping_address": {
                "zip": "10001"
            },
            "tos_acceptance": {
                "date": "2023-01-01",
                "ip": "192.168.1.1"
            }
        }
    ]
    for order in orders:
        if order["order_id"] == order_id:
            return order
    return "No order found"

async def retrieve_payment_intent(payment_intent_id):
    """
    Retrieve a Stripe payment intent by ID.
    Returns the payment intent object on success or an empty dictionary on failure.
    """
    try:
        return stripe.PaymentIntent.retrieve(payment_intent_id)
    except stripe.error.StripeError as e:
        logger.error(f"Stripe error occurred while retrieving payment intent: {e}")
        return {}

async def process_dispute(payment_intent_id):
    """Retrieve and process dispute data for a given PaymentIntent."""
    disputes_list = stripe.Dispute.list(payment_intent=payment_intent_id)
    if not disputes_list.data:
        logger.warning("No dispute data found for PaymentIntent: %s", payment_intent_id)
        return None
    dispute_data = disputes_list.data[0]

    payment_intent = await retrieve_payment_intent(payment_intent_id)
    order_id = int(payment_intent.get("metadata").get("order_id"))
    order = get_order(order_id)
    relevant_data = {
        "dispute_id": dispute_data.get("id"),
        "amount": dispute_data.get("amount"),
        "due_by": dispute_data.get("evidence_details", {}).get("due_by"),
        "reason": dispute_data.get("reason"),
        "status": dispute_data.get("status"),
        "card_brand": dispute_data.get("payment_method_details", {}).get("card", {}).get("brand"),
        # ---
        "payment_intent": payment_intent.to_dict_recursive(),
        "order": order
    }
    event_str = json.dumps(relevant_data)

    if order.get("fulfillment_details") == "not_shipped":
        result = await accept_dispute(relevant_data)
        logger.info("WORKFLOW RESULT: %s", result)
        return relevant_data, result
    else:
        result = await Runner.run(investigator_agent, input=event_str)
        logger.info("WORKFLOW RESULT: %s", result.final_output)
        return relevant_data, result.final_output

payment = stripe.PaymentIntent.create(
  amount=2000,
  currency="usd",
  payment_method = "pm_card_createDisputeProductNotReceived",
  confirm=True,
  metadata={"order_id": "1234"},
  off_session=True,
  automatic_payment_methods={"enabled": True},
)

asyncio.run(process_dispute(payment.id))

Dispute Details:
- Dispute ID: dp_1RoQ87QuMzbiv2kfQ9P1cjlN
- Amount: 2000
- Reason for Dispute: product not received

Order Details:
- Fulfillment status of the order: not shipped

Reasoning for closing the dispute:
The dispute is being closed as the product was not shipped or
fulfilled, and the customer did not receive their order. Since there
is no evidence of delivery or shipment, the refund is being issued and
the dispute finalized.

That's all I have for today! Talk to you soon ;)