How to Build an AI Agent with the OpenAI Agents SDK

1"""Billing/support triage agent: handoffs between specialists, guarded on input and output.
2Run: python billing_triage_agent.py (pip install openai-agents; set OPENAI_API_KEY)
3"""
4import asyncio
5import re
6
7from agents import (
8    Agent,
9    GuardrailFunctionOutput,
10    InputGuardrailTripwireTriggered,
11    OutputGuardrailTripwireTriggered,
12    RunContextWrapper,
13    Runner,
14    function_tool,
15    input_guardrail,
16    output_guardrail,
17)
18
19OFF_TOPIC_MARKERS = ("competitor", "unrelated to our product")
20
21
22@function_tool
23def look_up_invoice(customer_id: str) -> dict:
24    """Look up a customer's most recent invoice and payment status."""
25    return {"customer_id": customer_id, "amount": 49.00, "status": "paid", "invoice_id": "INV-2291"}
26
27
28@function_tool
29def check_known_issue(product_area: str) -> str:
30    """Check whether product_area has a known, already-tracked issue."""
31    return "No known issue for this area; treat as a new report."
32
33
34billing_agent = Agent(
35    name="Billing agent",
36    handoff_description="Handles invoices, payments, and refund requests.",
37    instructions="Help with billing questions. Look up the invoice before discussing any amount.",
38    tools=[look_up_invoice],
39)
40
41support_agent = Agent(
42    name="Technical agent",
43    handoff_description="Handles product bugs and how-to questions.",
44    instructions="Help with technical issues. Check for a known issue before answering.",
45    tools=[check_known_issue],
46)
47
48
49@input_guardrail
50async def reject_off_topic(ctx: RunContextWrapper[None], agent: Agent, input_text) -> GuardrailFunctionOutput:
51    text = input_text if isinstance(input_text, str) else str(input_text)
52    is_off_topic = any(marker in text.lower() for marker in OFF_TOPIC_MARKERS)
53    return GuardrailFunctionOutput(output_info={"off_topic": is_off_topic}, tripwire_triggered=is_off_topic)
54
55
56@output_guardrail
57async def block_large_refund_promise(ctx: RunContextWrapper[None], agent: Agent, output) -> GuardrailFunctionOutput:
58    text = output if isinstance(output, str) else getattr(output, "final_output", str(output))
59    match = re.search(r"refund.{0,20}\$(\d+(?:\.\d{2})?)", text, re.IGNORECASE)
60    too_large = bool(match) and float(match.group(1)) > 500
61    return GuardrailFunctionOutput(output_info={"amount": match.group(1) if match else None}, tripwire_triggered=too_large)
62
63
64billing_agent.output_guardrails = [block_large_refund_promise]
65
66triage_agent = Agent(
67    name="Triage agent",
68    instructions=(
69        "You are the first line of support for a SaaS product. Read the customer's "
70        "message and hand off to Billing agent for payment/invoice questions or "
71        "Technical agent for product issues. Don't try to answer either kind yourself."
72    ),
73    handoffs=[billing_agent, support_agent],
74    input_guardrails=[reject_off_topic],
75)
76
77
78async def handle_message(text: str) -> str:
79    try:
80        result = await Runner.run(triage_agent, text)
81        return result.final_output
82    except InputGuardrailTripwireTriggered:
83        return "This looks outside what we can help with here; try our public help center."
84    except OutputGuardrailTripwireTriggered:
85        return "[Routed to human review: proposed refund exceeds auto-approval limit]"
86
87
88if __name__ == "__main__":
89    print(asyncio.run(handle_message("My invoice looks wrong, can I get a refund?")))
90