Generative AIBuilding with foundation modelsTools & frameworksOpenAI Agents SDK

Advanced patterns in the OpenAI Agents SDK

9 minutes read

With the core building blocks in place—agents, tools, and tracing—the coding assistant works, but it isn't ready for production. We need to share state across agents, add human checkpoints to verify risky actions, and add guardrails to catch bad input or output before they cause harm. We also need session management to preserve conversation history across interactions. In this topic, we'll add these patterns to the coding assistant to make it more robust for production use.

Context management

You may have several agents calling multiple tools in your application. You need a way to share state across these agents and tools. To enable this, the Agents SDK provides context variables—a typed object you define and pass into the run. The SDK doesn't send this object to LLMs, but every agent and tool in that run can access it. In our coding assistant, for example, we can store state information such as:

  • The plan.

  • Code files.

  • Review comments.

Here is how we define the context object:

from dataclasses import dataclass, field

@dataclass
class ProjectContext:
    task_description: str
    plan_steps: list[str] = field(default_factory=list)

Then, we make this context available to tools and agents:

from agents import Agent, Runner, RunContextWrapper, function_tool

@function_tool
def save_plan(ctx: RunContextWrapper[ProjectContext], steps: list[str]) -> str:
    """Save the current plan steps to the project context."""
    ctx.context.plan_steps = steps
    return f"Saved {len(steps)} plan steps."

planner_agent = Agent[ProjectContext](
    name="Planner",
    model="gpt-5.4-mini",
    instructions="Break the task into steps. Use save_plan to store them.",
    tools=[save_plan],
)

context = ProjectContext(task_description="Build an email validator")
result = Runner.run_sync(planner_agent, "Build an email validator", context=context)

print(f"Plan: {context.plan_steps}")

The RunContextWrapper automatically injects as the first argument to any tool function that declares it. The SDK does not expose it to the model as a parameter. This gives tools read/write access to shared state without passing data through conversation messages.

Human-in-the-loop

Some decisions require human input. The SDK supports human-in-the-loop workflows, letting you pause the agent loop pending approval and resume when approved. State management becomes key here. But first, you must specify that a tool needs approval:

from agent import Agent, Runner, function_tool

@function_tool(needs_approval=True) # approval is needed to save files
def write_file(filename: str, content: str) -> str:
    """Write content to a file and return confirmation."""
    with open(filename, "w") as f:
        f.write(content)
    return f"Successfully wrote {len(content)} characters to {filename}"

coder_agent = Agent(
    name="Coder",
    model="gpt-5.4-mini",
    instructions="""You are a Python developer. Write clean, well-documented code.
    Use the write_file tool to save your code to files. If saving is not approved, politely ask how you can improve it.
    """,
    tools=[write_file],
)

result = Runner.run_sync(coder_agent, "Write a function that checks if a number is prime and save it to prime.py")

When approval is required, the agent loop is interrupted. You can check for interruptions and request approval or rejection:

while result.interruptions:
    state = result.to_state() # persist current state

    for interruption in result.interruptions:
        print(f"Tool {interruption.name} with args {interruption.arguments} is requesting approval.")
        user_input = input("Approve this action? (yes/no): ")

        if user_input.lower == "yes" or user_input.lower() == "y":
            state.approve(interruption, always_approve=False)
        else:
            state.reject(interruption)

    result = Runner.run_sync(coder_agent, state)

print(result.final_output)

For MCP server tools, approvals are specified with require_approval.

For long-running sessions, you can store state in a file or database with state.to_json() or state.to_string(), then restore it later with RunState.from_json() or RunState.from_string().

Session management

Sessions allow you to store conversation history without manual memory management. You can use various backing stores for this or use other mechanisms, such as the OpenAI Conversations API, previous_response_id, or auto_previous_response_id. However, you cannot use all of these approaches; you can only choose one.

For our coding assistant, we'll use a Redis data store:

# install package
# pip install openai-agents[redis]

from agents import Agent, Runner
from agents.extensions.memory import RedisSession
from dotenv import load_dotenv

load_dotenv()

redis_session = RedisSession.from_url(
    "hyper-user",
    url="redis://localhost:6379",
)

planner_agent = Agent(
    name="Planner",
    model="gpt-5.4-mini",
    instructions="""
    Break a coding task into steps. Never write any code just output the plan.
    """,
)

result = Runner.run_sync(
    planner_agent,
    "Hey, what can you help me with?",
    session=redis_session
)

follow_up = Runner.run_sync(
    planner_agent,
    "What did you say?",
    session=redis_session
)

print(follow_up.final_output)

You can run Redis via Docker with the following command:

docker run -d --name my-redis -p 6379:6379 redis:latest

You can perform various operations on a session, like below:

async def main():
    # view all items
    items = await redis_session.get_items()
    
    for item in items:
        print(item)
        
    # add new items
    new_item = [
        {"role": "user", "content": "Respond in all caps!"}
    ]
    
    await redis_session.add_items(new_item)
    
    # pop the most recent item
    recent_item = await redis_session.pop_item()
    print(recent_item)
    
    # clear all items
    await redis_session.clear_session()
    
if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

Besides Redis, you can also use:

  • SQLiteSession — a file- or in-memory backed store.

  • SQLAlchemySession — for databases that support SQLAlchemy.

  • DaprSession — for cloud-native deployments with Dapr sidecars.

  • OpenAIConversationsSession — sessions backed by the OpenAI Conversations API.

  • OpenAIResponsesCompactionSession — a wrapper around a session that enables automatic compaction when conversations get too long.

  • EncryptedSession — a wrapper around a session that enables encryption and TTL.

Guardrails

Guardrails run in parallel with the agent and can halt execution if something isn't right. They can validate the input before the agent processes it or the output before it returns. This is essential for safety, compliance, and quality control. For our coding assistant, we can set up a guardrail to ensure the user's input is coding-related.

First, we'll create a Pydantic object for structured output:

from pydantic import BaseModel

class RelevanceCheck(BaseModel):
    is_irrelevant: bool
    reasoning: str

Then, we need an agent to validate the user's input:

from agents import Agent

relevance_detector = Agent(
    name="Relevance Detector",
    model="gpt-5.4-mini",
    instructions=(
        "Determine if the input is irrelevant to a coding task. "
        "Irrelevant means the response is completely off-topic (e.g. poems, recipes, opinions) "
        "and not related to software development."
    ),
    output_type=RelevanceCheck,
)

Then, we'll create a guardrail function that returns a GuardrailFunctionOutput result:

from agents import (
    Runner,
    GuardrailFunctionOutput,
    RunContextWrapper,
    input_guardrail,
)

@input_guardrail
async def relevance_guardrail(
        ctx: RunContextWrapper, agent: Agent, output: str
) -> GuardrailFunctionOutput:
    verdict = await Runner.run(relevance_detector, str(output), context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=verdict.final_output,
        tripwire_triggered=verdict.final_output.is_irrelevant,
    )

Finally, we equip the agent with this guardrail:

planner_agent = Agent(
    name="Planner",
    model="gpt-5.4-mini",
    instructions="""
    You are a planning assistant. Break the coding task into a numbered step-by-step plan, then hand off to the Coder to implement it.
    """,
    input_guardrails=[relevance_guardrail],
)

If the guardrail is triggered, an exception InputGuardrailTripwireTriggered is raised. We can catch it and display relevant information to a user:

from agents import InputGuardrailTripwireTriggered

try:
    # irrelevant input
    result = Runner.run_sync(planner_agent, "I need to pick out shoes for my birthday.") 
    print(result.final_output)

except InputGuardrailTripwireTriggered:
    print("Only questions related to software development are allowed.")

Output guardrails work in a similar fashion. Your application can then reject malicious input, ask the user to rephrase, or flag outputs for human review.

Conclusion

These patterns move the coding assistant from a working prototype to a production-ready application. Here's a recap of what we covered:

  • Context management for sharing state across agents and tools without passing data through conversation messages.

  • Human-in-the-loop workflows that pause execution for human approval.

  • Session management to preserve conversation history across interactions, so users can pick up where they left off.

  • Guardrails to halt execution when safety or quality checks fail.

2 learners liked this piece of theory. 0 didn't like it. What about you?
Report a typo