scaffolding-openai-agents

name

scaffolding-openai-agents

description

Builds AI agents using OpenAI Agents SDK with async/await patterns and multi-agent orchestration. Use when creating tutoring agents, building agent handoffs, implementing tool-calling agents, or orchestrating multiple specialists. Covers Agent class, Runner patterns, function tools, guardrails, and streaming responses. Includes OmniChat distributed patterns: LLM Gateway routing, Persistence bridge, Event Publisher integration, and remote tool stubs. NOT when using raw OpenAI API without SDK or other agent frameworks like LangChain.

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_guardrail
async def check_homework_topic(context, agent, input_text: str) -> GuardrailFunctionOutput:
    """Ensure questions are homework-related."""
    keywords = ["python", "code", "programming", "function", "class", "error"]

    if not any(kw in input_text.lower() for kw in keywords):
        return GuardrailFunctionOutput(
            output_info="Not a programming question",
            tripwire_triggered=True
        )

    return GuardrailFunctionOutput(
        output_info="Valid programming question",
        tripwire_triggered=False
    )

tutor = Agent(
    name="Python Tutor",
    instructions="Help with Python homework.",
    input_guardrails=[check_homework_topic]
)

Output Validation

from agents import Agent, output_guardrail, GuardrailFunctionOutput

@output_guardrail
async def check_no_solutions(context, agent, output: str) -> GuardrailFunctionOutput:
    """Ensure we don't give complete homework solutions."""
    solution_indicators = ["here's the complete", "full solution", "copy this code"]

    if any(ind in output.lower() for ind in solution_indicators):
        return GuardrailFunctionOutput(
            output_info="Contains complete solution",
            tripwire_triggered=True
        )

    return GuardrailFunctionOutput(
        output_info="Output is appropriate",
        tripwire_triggered=False
    )

tutor = Agent(
    name="Python Tutor",
    instructions="Guide students without giving full solutions.",
    output_guardrails=[check_no_solutions]
)

Context Injection

Shared State Across Agents

from dataclasses import dataclass
from agents import Agent, Runner, function_tool, RunContextWrapper

@dataclass
class TutoringContext:
    student_id: str
    session_id: str
    topics_covered: list[str]
    difficulty_level: str = "beginner"

@function_tool
def log_topic(wrapper: RunContextWrapper[TutoringContext], topic: str) -> str:
    """Log a topic as covered in this session."""
    wrapper.context.topics_covered.append(topic)
    return f"Logged: {topic}"

tutor = Agent(
    name="Python Tutor",
    instructions="Teach Python, tracking topics covered.",
    tools=[log_topic]
)

async def main():
    ctx = TutoringContext(
        student_id="student-123",
        session_id="session-456",
        topics_covered=[]
    )

    result = await Runner.run(
        tutor,
        "Teach me about loops",
        context=ctx
    )

    print(f"Topics covered: {ctx.topics_covered}")

Project Structure

learnflow-agents/
├── agents/
│   ├── __init__.py
│   ├── triage.py          # Routing agent
│   ├── concepts.py        # Concepts specialist
│   ├── debug.py           # Debug specialist
│   └── exercise.py        # Exercise generator
├── tools/
│   ├── __init__.py
│   ├── code_runner.py     # Execute Python safely
│   └── search.py          # Search documentation
├── guardrails/
│   ├── __init__.py
│   ├── input.py           # Input validation
│   └── output.py          # Output validation
├── main.py                # FastAPI integration
└── pyproject.toml

FastAPI Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agents import Agent, Runner

app = FastAPI()

# Initialize agents
triage = Agent(
    name="Triage",
    instructions="Route questions to specialists",
    handoffs=[concepts_agent, debug_agent]
)

class Question(BaseModel):
    text: str
    session_id: str

class Answer(BaseModel):
    response: str
    agent_used: str

@app.post("/ask", response_model=Answer)
async def ask_question(question: Question):
    try:
        result = await Runner.run(triage, question.text)
        return Answer(
            response=result.final_output,
            agent_used=result.last_agent.name
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/ask/stream")
async def ask_stream(question: Question):
    from fastapi.responses import StreamingResponse

    async def generate():
        result = Runner.run_streamed(triage, question.text)
        async for event in result.stream_events():
            if hasattr(event, 'delta'):
                yield event.delta

    return StreamingResponse(generate(), media_type="text/plain")

OmniChat Distributed Patterns

The following patterns adapt the OpenAI Agents SDK for the OmniChat microservices architecture.

1. LLM Gateway Routing

The Orchestrator routes all LLM calls through the internal LLM Gateway (Phase 6), not directly to OpenAI.

# config.py
from pydantic_settings import BaseSettings

class OrchestratorSettings(BaseSettings):
    llm_gateway_url: str = "http://llm-gateway:8000/v1"
    llm_gateway_timeout: int = 120

    model_config = {"env_prefix": "ORCHESTRATOR_"}

settings = OrchestratorSettings()

# agents/base.py
from agents import Agent, ModelSettings
from openai import AsyncOpenAI

# Create client pointing to LLM Gateway
llm_client = AsyncOpenAI(
    base_url=settings.llm_gateway_url,
    api_key="internal",  # Gateway handles real keys
    default_headers={
        "X-Service": "chat-orchestrator",
        "X-Provider": "openai",  # or "anthropic"
    },
    timeout=settings.llm_gateway_timeout,
)

def create_agent(
    name: str,
    instructions: str,
    model: str = "gpt-4o",
    **kwargs
) -> Agent:
    """Factory for agents using LLM Gateway."""
    return Agent(
        name=name,
        instructions=instructions,
        model=model,
        model_settings=ModelSettings(
            openai_client=llm_client,
        ),
        **kwargs
    )

# Usage
triage_agent = create_agent(
    name="Triage",
    instructions="Route user requests to appropriate specialists.",
    handoffs=[search_agent, code_agent, chat_agent],
)

2. Stateless Persistence Bridge

Fetch conversation history from Persistence Service at each turn start.

# services/persistence.py
import httpx
from pydantic import BaseModel

class Message(BaseModel):
    role: str
    content: str

class PersistenceClient:
    def __init__(self, base_url: str = "http://persistence-service:8000"):
        self.base_url = base_url
        self._client = httpx.AsyncClient(base_url=base_url, timeout=10)

    async def get_conversation_history(
        self,
        conversation_id: str,
        limit: int = 50,
    ) -> list[dict]:
        """Fetch recent messages for Runner input."""
        response = await self._client.get(
            f"/conversations/{conversation_id}/messages",
            params={"limit": limit, "order": "asc"},
        )
        response.raise_for_status()

        messages = response.json()["items"]
        return [
            {"role": msg["role"], "content": msg["content"]}
            for msg in messages
        ]

    async def save_message(
        self,
        conversation_id: str,
        role: str,
        content: str,
        metadata: dict | None = None,
    ) -> str:
        """Persist a message and return its ID."""
        response = await self._client.post(
            f"/conversations/{conversation_id}/messages",
            json={"role": role, "content": content, "metadata": metadata or {}},
        )
        response.raise_for_status()
        return response.json()["id"]

persistence = PersistenceClient()

# handlers/chat.py
async def handle_chat_turn(
    conversation_id: str,
    user_id: str,
    user_message: str,
) -> str:
    """Process a single chat turn with persistence."""

    # 1. Fetch history from Persistence Service
    history = await persistence.get_conversation_history(conversation_id)

    # 2. Append new user message
    input_messages = history + [{"role": "user", "content": user_message}]

    # 3. Save user message
    await persistence.save_message(conversation_id, "user", user_message)

    # 4. Run agent
    result = await Runner.run(triage_agent, input_messages)

    # 5. Save assistant response
    await persistence.save_message(
        conversation_id,
        "assistant",
        result.final_output,
        metadata={"agent": result.last_agent.name},
    )

    return result.final_output

3. Event Publisher Integration

Push real-time status updates and tokens to Stream Broker via Redis.

# services/events.py
from events.publisher import EventPublisher
import redis.asyncio as redis

redis_client = redis.Redis.from_url(settings.redis_url)
publisher = EventPublisher(redis_client)

# handlers/stream.py
from agents import Agent, Runner
from agents.stream_events import (
    RawResponsesStreamEvent,
    RunItemStreamEvent,
)

async def handle_chat_stream(
    conversation_id: str,
    user_id: str,
    user_message: str,
):
    """Process chat with real-time streaming to Stream Broker."""

    # Fetch history
    history = await persistence.get_conversation_history(conversation_id)
    input_messages = history + [{"role": "user", "content": user_message}]

    # Generate message ID for this response
    message_id = str(uuid.uuid4())

    # Publish initial status
    await publisher.status(user_id, "thinking", "Processing your request...")

    # Run with streaming
    stream_result = Runner.run_streamed(triage_agent, input_messages)
    collected_content = []

    async for event in stream_result.stream_events():
        # Token deltas
        if isinstance(event, RawResponsesStreamEvent):
            if hasattr(event.data, "delta") and event.data.delta.content:
                for content_part in event.data.delta.content:
                    if hasattr(content_part, "text"):
                        token = content_part.text
                        collected_content.append(token)
                        await publisher.delta(user_id, token, message_id)

        # Tool calls
        elif isinstance(event, RunItemStreamEvent):
            item = event.item
            if hasattr(item, "type"):
                if item.type == "tool_call_item":
                    await publisher.tool_start(
                        user_id,
                        item.tool_call.name,
                        item.tool_call.id,
                    )
                elif item.type == "tool_call_output_item":
                    await publisher.tool_end(
                        user_id,
                        item.tool_call.name,
                        item.tool_call.id,
                        {"output": item.output[:500]},  # Truncate
                    )

    # Get final result
    result = await stream_result.result()

    # Persist final response
    final_content = "".join(collected_content) or result.final_output
    await persistence.save_message(
        conversation_id,
        "assistant",
        final_content,
        metadata={"agent": result.last_agent.name, "message_id": message_id},
    )

    return final_content

4. Remote Tool Stubs (arq Task Queue)

Tools dispatch work to Tool Executor service and await results.

# tools/stubs.py
"""
Remote tool stubs - dispatch to Tool Executor via arq queue.
The Orchestrator defines WHAT to do; Tool Executor defines HOW.
"""
from agents import function_tool
from arq import ArqRedis
from arq.connections import RedisSettings
import asyncio

# arq connection
arq_redis: ArqRedis | None = None

async def get_arq() -> ArqRedis:
    global arq_redis
    if arq_redis is None:
        arq_redis = await ArqRedis.from_url(settings.redis_url)
    return arq_redis


@function_tool
async def web_search(query: str, max_results: int = 5) -> str:
    """Search the web for information.

    Args:
        query: The search query
        max_results: Maximum number of results to return
    """
    arq = await get_arq()

    # Dispatch to Tool Executor
    job = await arq.enqueue_job(
        "execute_web_search",  # Task name in Tool Executor
        query,
        max_results,
        _queue_name="tools:search",
    )

    # Await result (with timeout)
    try:
        result = await asyncio.wait_for(job.result(), timeout=30)
        return result
    except asyncio.TimeoutError:
        return "Search timed out. Please try again."


@function_tool
async def run_python_code(code: str) -> str:
    """Execute Python code in a secure sandbox.

    Args:
        code: Python code to execute
    """
    arq = await get_arq()

    job = await arq.enqueue_job(
        "execute_python",
        code,
        _queue_name="tools:code",
    )

    try:
        result = await asyncio.wait_for(job.result(), timeout=60)
        return f"Output:\n{result.get('stdout', '')}\n{result.get('stderr', '')}"
    except asyncio.TimeoutError:
        return "Code execution timed out."


@function_tool
async def generate_image(prompt: str, style: str = "realistic") -> str:
    """Generate an image from a text prompt.

    Args:
        prompt: Description of the image to generate
        style: Image style (realistic, cartoon, sketch)
    """
    arq = await get_arq()

    job = await arq.enqueue_job(
        "generate_image",
        prompt,
        style,
        _queue_name="tools:media",
    )

    try:
        result = await asyncio.wait_for(job.result(), timeout=120)
        return f"Image generated: {result['url']}"
    except asyncio.TimeoutError:
        return "Image generation timed out."

# agents/specialist.py
"""Specialist agents with remote tool stubs."""
from .base import create_agent
from tools.stubs import web_search, run_python_code, generate_image

search_agent = create_agent(
    name="Search Specialist",
    instructions="""You help users find information on the web.
    Use the web_search tool to find relevant results.
    Summarize findings clearly with sources.""",
    handoff_description="Searches the web for information",
    tools=[web_search],
)

code_agent = create_agent(
    name="Code Specialist",
    instructions="""You help users write and run Python code.
    Use run_python_code to execute code in a sandbox.
    Explain outputs and fix errors.""",
    handoff_description="Writes and executes Python code",
    tools=[run_python_code],
)

media_agent = create_agent(
    name="Media Specialist",
    instructions="""You help users create images.
    Use generate_image with detailed prompts.
    Describe what you're creating before generating.""",
    handoff_description="Generates images from descriptions",
    tools=[generate_image],
)

Full Orchestrator Service Structure

services/chat-orchestrator/
├── src/chat_orchestrator/
│   ├── __init__.py
│   ├── main.py              # FastAPI app
│   ├── config.py            # Settings
│   ├── agents/
│   │   ├── __init__.py
│   │   ├── base.py          # create_agent factory
│   │   ├── triage.py        # Routing agent
│   │   └── specialists.py   # Search, Code, Media agents
│   ├── tools/
│   │   ├── __init__.py
│   │   └── stubs.py         # Remote tool stubs
│   ├── services/
│   │   ├── __init__.py
│   │   ├── persistence.py   # Persistence client
│   │   └── events.py        # Event publisher
│   ├── handlers/
│   │   ├── __init__.py
│   │   ├── chat.py          # Non-streaming handler
│   │   └── stream.py        # Streaming handler
│   └── guardrails/
│       ├── __init__.py
│       └── content.py       # Input/output validation
├── tests/
├── pyproject.toml
└── Dockerfile

Skill Relationships

Concern	Skill	How They Connect
Real-time delivery	streaming-llm-responses	Orchestrator publishes → Stream Broker delivers via SSE
Event publishing	event-publisher	Orchestrator uses EventPublisher to push status/tokens
Task queuing	redis-pattern	Tools dispatch to arq queues for Tool Executor
Data access	scaffolding-fastapi-sqlmodel	Persistence Service exposes REST API

Tracing & Debugging

View Traces

Traces available at: https://platform.openai.com/traces

Custom Tracing

from agents import Runner, RunConfig

config = RunConfig(
    workflow_name="tutoring-session",
    trace_id="custom-trace-123"
)

result = await Runner.run(agent, "Hello", run_config=config)

Verification

Run: python scripts/verify.py

Related Skills

Skill	Relationship
event-publisher	Producer API for pushing events to Redis
streaming-llm-responses	Consumer (Stream Broker) that delivers events via SSE
redis-pattern	arq task queue patterns for remote tool dispatch
scaffolding-fastapi-sqlmodel	Persistence Service structure
adapter-interface	LLM Gateway provider abstraction