marketplace

Scaffolding OpenAI Agents

Build production AI agents using OpenAI Agents SDK with native async/await patterns.

Quick Start

# Project setup
mkdir my-agent && cd my-agent
python -m venv .venv && source .venv/bin/activate
pip install openai-agents

# Set API key
export OPENAI_API_KEY=sk-...

# main.py
import asyncio
from agents import Agent, Runner

agent = Agent(
    name="Python Tutor",
    instructions="You help students learn Python. Explain concepts clearly with examples."
)

async def main():
    result = await Runner.run(agent, "Explain list comprehensions")
    print(result.final_output)

asyncio.run(main())

Agent Configuration

Basic Agent

from agents import Agent

tutor = Agent(
    name="Python Tutor",
    instructions="""You are an expert Python tutor.
    Explain concepts clearly with examples.
    Ask clarifying questions when needed.
    Provide practice exercises after explanations.""",
    model="gpt-4o"
)

With Model Settings

from agents import Agent, ModelSettings

agent = Agent(
    name="Creative Writer",
    instructions="Write creative stories based on prompts.",
    model="gpt-4o",
    model_settings=ModelSettings(
        temperature=0.9,
        max_tokens=2000
    )
)

With Structured Output

from pydantic import BaseModel
from agents import Agent

class CodeReview(BaseModel):
    issues: list[str]
    suggestions: list[str]
    score: int

reviewer = Agent(
    name="Code Reviewer",
    instructions="Review Python code for issues and improvements.",
    output_type=CodeReview  # Forces structured JSON output
)

Runner Patterns

Async Run (Primary)

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(name="Helper", instructions="Be helpful")

    # Single query
    result = await Runner.run(agent, "What is Python?")
    print(result.final_output)

    # With conversation history
    messages = [
        {"role": "user", "content": "My name is Alex"},
        {"role": "assistant", "content": "Nice to meet you, Alex!"},
        {"role": "user", "content": "What's my name?"}
    ]
    result = await Runner.run(agent, messages)
    print(result.final_output)  # "Your name is Alex"

asyncio.run(main())

Sync Run (Simple Scripts)

from agents import Agent, Runner

agent = Agent(name="Helper", instructions="Be helpful")
result = Runner.run_sync(agent, "Hello!")
print(result.final_output)

Streaming Run

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(name="Storyteller", instructions="Tell engaging stories")

    result = Runner.run_streamed(agent, "Tell me a short story")

    async for event in result.stream_events():
        if hasattr(event, 'delta'):
            print(event.delta, end='', flush=True)

    print()  # Newline at end

asyncio.run(main())

Conversation Continuation

async def chat_session():
    agent = Agent(name="Tutor", instructions="You are a Python tutor")

    # First turn
    result1 = await Runner.run(agent, "Explain decorators")
    print(f"Tutor: {result1.final_output}")

    # Continue conversation
    messages = result1.to_input_list() + [
        {"role": "user", "content": "Show me an example"}
    ]
    result2 = await Runner.run(agent, messages)
    print(f"Tutor: {result2.final_output}")

Function Tools

Basic Tool

from agents import Agent, function_tool

@function_tool
def get_current_time() -> str:
    """Get the current time."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S")

@function_tool
def calculate(expression: str) -> float:
    """Calculate a mathematical expression.

    Args:
        expression: A valid Python math expression like "2 + 2" or "10 * 5"
    """
    return eval(expression)  # Use safe_eval in production

agent = Agent(
    name="Assistant",
    instructions="Help with calculations and time queries.",
    tools=[get_current_time, calculate]
)

Async Tool

import httpx
from agents import Agent, function_tool

@function_tool
async def fetch_weather(city: str) -> str:
    """Fetch current weather for a city.

    Args:
        city: The city name to get weather for
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}?format=3"
        )
        return response.text

agent = Agent(
    name="Weather Bot",
    instructions="Provide weather information.",
    tools=[fetch_weather]
)

Tool with Pydantic Types

from pydantic import BaseModel
from agents import Agent, function_tool

class SearchQuery(BaseModel):
    query: str
    max_results: int = 10

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str

@function_tool
async def search_docs(params: SearchQuery) -> list[SearchResult]:
    """Search documentation for a query."""
    # Implementation
    return [SearchResult(
        title="Python Tutorial",
        url="https://docs.python.org",
        snippet="Official Python documentation..."
    )]

agent = Agent(
    name="Doc Search",
    instructions="Search Python documentation.",
    tools=[search_docs]
)

Multi-Agent Patterns

Handoffs (Recommended for Routing)

from agents import Agent, Runner

# Specialist agents
concepts_agent = Agent(
    name="Concepts Tutor",
    handoff_description="Explains Python concepts and fundamentals",
    instructions="Explain Python concepts clearly with examples."
)

debug_agent = Agent(
    name="Debug Helper",
    handoff_description="Helps debug Python code errors",
    instructions="Help diagnose and fix Python errors."
)

exercise_agent = Agent(
    name="Exercise Generator",
    handoff_description="Creates practice problems and exercises",
    instructions="Generate practice problems with solutions."
)

# Triage agent with handoffs
triage_agent = Agent(
    name="Triage",
    instructions="""Route student questions to the right specialist:
    - Concepts questions → Concepts Tutor
    - Error/bug questions → Debug Helper
    - Practice requests → Exercise Generator

    Analyze the question and hand off to the appropriate agent.""",
    handoffs=[concepts_agent, debug_agent, exercise_agent]
)

async def main():
    # Question gets routed automatically
    result = await Runner.run(
        triage_agent,
        "I'm getting a KeyError in my dictionary code"
    )
    print(result.final_output)  # Handled by debug_agent

Agents as Tools (Orchestration)

from agents import Agent, Runner

# Create specialist agents
researcher = Agent(
    name="Researcher",
    instructions="Research topics thoroughly."
)

writer = Agent(
    name="Writer",
    instructions="Write clear, engaging content."
)

# Manager uses agents as tools
manager = Agent(
    name="Content Manager",
    instructions="""Coordinate research and writing:
    1. Use researcher tool to gather information
    2. Use writer tool to create content""",
    tools=[
        researcher.as_tool(
            tool_name="research",
            tool_description="Research a topic"
        ),
        writer.as_tool(
            tool_name="write",
            tool_description="Write content about a topic"
        )
    ]
)

async def main():
    result = await Runner.run(
        manager,
        "Create a blog post about async Python"
    )
    print(result.final_output)

Guardrails

Input Validation

from agents import Agent, input_guardrail, GuardrailFunctionOutput

@input_g