AgentKit Walkthrough

Build production-ready AI agents using OpenAI's official agent framework with tools, guardrails, and tracing.

Overview

OpenAI's Agents SDK (AgentKit) provides a structured framework for building agents with function calling, handoffs, and observability built-in.

┌─────────────────────────────────────────────────────────────────────┐
│                    AGENTKIT ARCHITECTURE                             │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │                         AGENT                                │    │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │    │
│  │  │ Instructions │  │    Model     │  │    Tools     │       │    │
│  │  │   (Prompt)   │  │  (gpt-4o)    │  │  (Functions) │       │    │
│  │  └──────────────┘  └──────────────┘  └──────────────┘       │    │
│  │                                                              │    │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │    │
│  │  │  Guardrails  │  │   Handoffs   │  │   Context    │       │    │
│  │  │  (Input/Out) │  │ (to Agents)  │  │  Variables   │       │    │
│  │  └──────────────┘  └──────────────┘  └──────────────┘       │    │
│  └─────────────────────────────────────────────────────────────┘    │
│                             │                                        │
│                             ▼                                        │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │                        RUNNER                                │    │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │    │
│  │  │   Execute    │  │    Trace     │  │   Manage     │       │    │
│  │  │    Turns     │  │   Logging    │  │    State     │       │    │
│  │  └──────────────┘  └──────────────┘  └──────────────┘       │    │
│  └─────────────────────────────────────────────────────────────┘    │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Installation

pip install openai-agents

Basic Agent

from agents import Agent, Runner

# Create a simple agent
agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant that answers questions clearly and concisely.",
    model="gpt-4o"
)

# Run the agent
result = Runner.run_sync(agent, "What is the capital of France?")
print(result.final_output)

Adding Tools

Function Tools

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a city.

    Args:
        city: The name of the city
    """
    # In production, call a real weather API
    return f"The weather in {city} is sunny and 72°F"

@function_tool
def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the product database.

    Args:
        query: Search query string
        limit: Maximum number of results to return
    """
    # In production, query your database
    return [{"id": "1", "name": "Product A", "price": 29.99}]

# Create agent with tools
agent = Agent(
    name="Sales Assistant",
    instructions="""You are a sales assistant that helps customers find products.
    Use the search_database tool to find products.
    Use get_weather for small talk if customers ask about weather.""",
    tools=[get_weather, search_database],
    model="gpt-4o"
)

result = Runner.run_sync(agent, "Do you have any headphones?")
print(result.final_output)

Structured Output Tools

from pydantic import BaseModel
from agents import Agent, Runner, function_tool

class ProductRecommendation(BaseModel):
    product_name: str
    reason: str
    confidence: float
    price_range: str

@function_tool
def recommend_product(
    customer_needs: str,
    budget: float
) -> ProductRecommendation:
    """Recommend a product based on customer needs and budget.

    Args:
        customer_needs: Description of what the customer is looking for
        budget: Customer's maximum budget in USD
    """
    # In production, use ML model or rules engine
    return ProductRecommendation(
        product_name="Premium Headphones X1",
        reason="Matches your need for noise cancellation",
        confidence=0.92,
        price_range="$150-$200"
    )

agent = Agent(
    name="Product Advisor",
    instructions="Help customers find the perfect product. Use the recommend_product tool.",
    tools=[recommend_product],
    model="gpt-4o"
)

Agent Handoffs

Transfer conversations between specialized agents:

from agents import Agent, Runner

# Define specialized agents
sales_agent = Agent(
    name="Sales Agent",
    instructions="""You handle sales inquiries, product questions, and purchases.
    If customer has a technical issue, transfer to support.
    If customer has billing questions, transfer to billing.""",
    model="gpt-4o"
)

support_agent = Agent(
    name="Support Agent",
    instructions="""You handle technical support and troubleshooting.
    If customer wants to buy something, transfer to sales.
    If customer has billing questions, transfer to billing.""",
    model="gpt-4o"
)

billing_agent = Agent(
    name="Billing Agent",
    instructions="""You handle billing, invoices, and payment issues.
    If customer has product questions, transfer to sales.
    If customer has technical issues, transfer to support.""",
    model="gpt-4o"
)

# Set up handoffs
sales_agent.handoffs = [support_agent, billing_agent]
support_agent.handoffs = [sales_agent, billing_agent]
billing_agent.handoffs = [sales_agent, support_agent]

# Create triage agent
triage_agent = Agent(
    name="Triage Agent",
    instructions="""You are the first point of contact.
    Understand what the customer needs and transfer to the right specialist:
    - Product questions, pricing, purchases → Sales Agent
    - Technical problems, bugs, how-to → Support Agent
    - Invoice, payment, refund issues → Billing Agent""",
    handoffs=[sales_agent, support_agent, billing_agent],
    model="gpt-4o"
)

# Run with automatic handoffs
result = Runner.run_sync(triage_agent, "I can't log into my account")
print(f"Final agent: {result.last_agent.name}")
print(f"Response: {result.final_output}")

Guardrails

Input Guardrails

from agents import Agent, Runner, InputGuardrail, GuardrailResult
from pydantic import BaseModel

class TopicCheck(BaseModel):
    is_appropriate: bool
    reason: str

async def check_topic(input_text: str) -> GuardrailResult:
    """Check if input is appropriate for this agent."""

    # Use a fast model for guardrail checks
    check_agent = Agent(
        name="Topic Checker",
        instructions="""Determine if the input is appropriate for a customer service bot.
        Appropriate: product questions, support, billing
        Inappropriate: medical advice, legal advice, hate speech""",
        model="gpt-4o-mini",
        output_type=TopicCheck
    )

    result = await Runner.run(check_agent, input_text)
    topic_check = result.final_output_as(TopicCheck)

    if not topic_check.is_appropriate:
        return GuardrailResult(
            blocked=True,
            message=f"I can only help with customer service topics. {topic_check.reason}"
        )

    return GuardrailResult(blocked=False)

# Apply guardrail
agent = Agent(
    name="Customer Service",
    instructions="You help customers with product and service questions.",
    input_guardrails=[InputGuardrail(check_topic)],
    model="gpt-4o"
)

Output Guardrails

from agents import Agent, OutputGuardrail, GuardrailResult

class QualityCheck(BaseModel):
    is_professional: bool
    issues: list[str]

async def check_output_quality(output_text: str) -> GuardrailResult:
    """Verify output meets quality standards."""

    check_agent = Agent(
        name="Quality Checker",
        instructions="""Check if the response is professional and appropriate.
        Flag: inappropriate language, incorrect info, rude tone""",
        model="gpt-4o-mini",
        output_type=QualityCheck
    )

    result = await Runner.run(check_agent, f"Check this response:\n\n{output_text}")
    quality = result.final_output_as(QualityCheck)

    if not quality.is_professional:
        return GuardrailResult(
            blocked=True,
            message="Response did not meet quality standards.",
            issues=quality.issues
        )

    return GuardrailResult(blocked=False)

agent = Agent(
    name="Support Agent",
    instructions="Help customers with technical issues.",
    output_guardrails=[OutputGuardrail(check_output_quality)],
    model="gpt-4o"
)

Context Variables

Pass and update context throughout the conversation:

from agents import Agent, Runner, RunContext
from typing import Any

@function_tool
def get_customer_info(ctx: RunContext) -> dict:
    """Get information about the current customer."""
    customer_id = ctx.context_variables.get("customer_id")
    # In production, fetch from database
    return {
        "id": customer_id,
        "name": "John Doe",
        "tier": "premium",
        "balance": 150.00
    }

@function_tool
def update_customer_notes(ctx: RunContext, note: str) -> str:
    """Add a note to the customer's account.

    Args:
        note: The note to add
    """
    customer_id = ctx.context_variables.get("customer_id")
    # Update the context with the note
    ctx.context_variables["notes"] = ctx.context_variables.get("notes", [])
    ctx.context_variables["notes"].append(note)
    return f"Note added for customer {customer_id}"

agent = Agent(
    name="Account Manager",
    instructions="""Help manage customer accounts.
    Always greet the customer by name after looking up their info.""",
    tools=[get_customer_info, update_customer_notes],
    model="gpt-4o"
)

# Run with initial context
result = Runner.run_sync(
    agent,
    "What's my account balance?",
    context_variables={"customer_id": "cust_12345"}
)

Tracing and Observability

from agents import Agent, Runner, Tracer
from agents.tracing import ConsoleTracer, FileTracer

# Console tracing (for development)
console_tracer = ConsoleTracer()

# File tracing (for production)
file_tracer = FileTracer(path="./traces")

agent = Agent(
    name="Traced Agent",
    instructions="You are a helpful assistant.",
    model="gpt-4o"
)

# Run with tracing enabled
result = Runner.run_sync(
    agent,
    "Hello!",
    tracer=console_tracer
)

# Traces show:
# - Agent turns
# - Tool calls and results
# - Handoffs
# - Token usage
# - Latency

Custom Tracing

from agents.tracing import Tracer, TraceEvent
from typing import Any

class CustomTracer(Tracer):
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.events = []

    def on_event(self, event: TraceEvent):
        # Log to your observability platform
        self.events.append({
            "service": self.service_name,
            "event_type": event.type,
            "agent": event.agent_name,
            "timestamp": event.timestamp,
            "data": event.data
        })

        # Send to external service (e.g., Datadog, New Relic)
        # self.send_to_datadog(event)

    def get_summary(self) -> dict:
        return {
            "total_events": len(self.events),
            "agents_used": list(set(e["agent"] for e in self.events)),
            "event_types": list(set(e["event_type"] for e in self.events))
        }

# Usage
tracer = CustomTracer("customer-service")
result = Runner.run_sync(agent, "Help me", tracer=tracer)
print(tracer.get_summary())

Async Execution

import asyncio
from agents import Agent, Runner

agent = Agent(
    name="Async Agent",
    instructions="You are helpful.",
    model="gpt-4o"
)

async def main():
    # Single async call
    result = await Runner.run(agent, "Hello!")
    print(result.final_output)

    # Parallel async calls
    tasks = [
        Runner.run(agent, f"Question {i}")
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)

    for i, result in enumerate(results):
        print(f"Result {i}: {result.final_output}")

asyncio.run(main())

Complete Example: Customer Service Bot

from agents import Agent, Runner, function_tool, InputGuardrail, GuardrailResult
from pydantic import BaseModel
from typing import Optional
import asyncio

# --- Data Models ---
class Order(BaseModel):
    id: str
    status: str
    items: list[str]
    total: float

class Customer(BaseModel):
    id: str
    name: str
    email: str
    tier: str

# --- Tools ---
@function_tool
def get_order_status(order_id: str) -> Order:
    """Look up an order by ID.

    Args:
        order_id: The order ID to look up
    """
    # In production, query database
    return Order(
        id=order_id,
        status="shipped",
        items=["Widget A", "Widget B"],
        total=99.99
    )

@function_tool
def initiate_refund(order_id: str, reason: str) -> dict:
    """Start a refund for an order.

    Args:
        order_id: The order to refund
        reason: Reason for the refund
    """
    return {
        "refund_id": "ref_123",
        "status": "processing",
        "estimated_days": 3
    }

@function_tool
def update_shipping_address(order_id: str, new_address: str) -> dict:
    """Update shipping address for an order.

    Args:
        order_id: The order to update
        new_address: The new shipping address
    """
    return {"status": "updated", "order_id": order_id}

# --- Guardrails ---
class InputCheck(BaseModel):
    is_customer_service: bool
    detected_intent: str

async def topic_guardrail(input_text: str) -> GuardrailResult:
    check_agent = Agent(
        name="Intent Classifier",
        instructions="Classify if this is a customer service request.",
        model="gpt-4o-mini",
        output_type=InputCheck
    )
    result = await Runner.run(check_agent, input_text)
    check = result.final_output_as(InputCheck)

    if not check.is_customer_service:
        return GuardrailResult(
            blocked=True,
            message="I'm a customer service bot. I can help with orders, refunds, and shipping."
        )
    return GuardrailResult(blocked=False)

# --- Agents ---
order_agent = Agent(
    name="Order Specialist",
    instructions="""You help with order-related inquiries.
    - Check order status
    - Update shipping addresses
    - Track packages

    For refunds, transfer to the refund specialist.""",
    tools=[get_order_status, update_shipping_address],
    model="gpt-4o"
)

refund_agent = Agent(
    name="Refund Specialist",
    instructions="""You handle refund requests.
    - Verify order eligibility
    - Process refunds
    - Explain refund policies

    Be empathetic but follow policy.""",
    tools=[get_order_status, initiate_refund],
    model="gpt-4o"
)

# Set up handoffs
order_agent.handoffs = [refund_agent]
refund_agent.handoffs = [order_agent]

# Main triage agent
triage_agent = Agent(
    name="Customer Service",
    instructions="""You are the main customer service agent.
    Route requests appropriately:
    - Order status, shipping → Order Specialist
    - Refunds, returns → Refund Specialist

    Be friendly and professional.""",
    handoffs=[order_agent, refund_agent],
    input_guardrails=[InputGuardrail(topic_guardrail)],
    model="gpt-4o"
)

# --- Run ---
async def main():
    # Test various scenarios
    test_cases = [
        "Where is my order #12345?",
        "I want to return my order",
        "Can you help me with my taxes?",  # Should be blocked
    ]

    for query in test_cases:
        print(f"\n{'='*50}")
        print(f"Query: {query}")
        try:
            result = await Runner.run(triage_agent, query)
            print(f"Agent: {result.last_agent.name}")
            print(f"Response: {result.final_output}")
        except Exception as e:
            print(f"Blocked: {e}")

asyncio.run(main())

See Also

• Orchestrating Agents
• LLM Guardrails
• Function Calling