InkdownInkdown
Start writing

OpenAI Agents Python

21 filesยท0 subfolders

Shared Workspace

OpenAI Agents Python
00_OVERVIEW.md

13_LIFECYCLE_HOOKS

Shared from "OpenAI Agents Python" on Inkdown

Lifecycle Hooks - Comprehensive Deep Dive

Overview

Lifecycle Hooks in the OpenAI Agents SDK allow you to execute custom code at specific points during an agent run. Think of Lifecycle Hooks as "event listeners" or "callbacks" that let you hook into the execution flow - you can run custom logic when an agent starts, when a tool is called, when the LLM generates a response, and more. This is essential for logging, monitoring, custom business logic, and extending agent behavior.

Core Concepts

What are Lifecycle Hooks?

Lifecycle Hooks are functions that are called at specific points during agent execution:

  • Agent lifecycle - When an agent starts or ends
  • LLM lifecycle - Before and after LLM calls
  • Tool lifecycle - Before and after tool execution
  • Handoff lifecycle - When agents hand off to each other
Why Lifecycle Hooks Matter
01_AGENT_SYSTEM.md
02_RUNNER_SYSTEM.md
03_TOOL_SYSTEM.md
04_ITEMS_SYSTEM.md
05_GUARDRAILS.md
06_HANDOFFS.md
07_MEMORY_SESSIONS.md
08_MODEL_PROVIDERS.md
09_SANDBOX_SYSTEM.md
10_TRACING.md
11_RUN_STATE.md
12_CONTEXT.md
13_LIFECYCLE_HOOKS.md
14_CONFIGURATION.md
15_ERROR_HANDLING.md
16_STREAMING.md
17_EXTENSIONS.md
18_MCP_INTEGRATION.md
19_BEST_PRACTICES.md
20_ARCHITECTURE_PATTERNS.md
  • Logging - Log events for monitoring and debugging
  • Monitoring - Track agent behavior and performance
  • Custom Logic - Implement business rules at specific points
  • Analytics - Collect data on agent usage
  • Validation - Add custom validation at key points
  • Extension - Extend agent behavior without modifying core code

    • Hook Types

    Run Hooks

    RunHooks are called for the entire run, across all agents:

    Python
    from agents import RunHooks

class MyRunHooks(RunHooks[TContext]):
    async def on_agent_start(self, context, agent):
        """Called when any agent starts."""
        print(f"Agent {agent.name} starting")
    
    async def on_agent_end(self, context, agent, output):
        """Called when any agent ends."""
        print(f"Agent {agent.name} finished")

    When to use:

    • Global logging for all agents
    • Cross-agent monitoring
    • Run-level analytics
    • Global business logic
    Agent Hooks

    AgentHooks are called for a specific agent:

    Python
    from agents import AgentHooks

class MyAgentHooks(AgentHooks[TContext]):
    async def on_start(self, context, agent):
        """Called when this specific agent starts."""
        print(f"My agent starting")
    
    async def on_end(self, context, agent, output):
        """Called when this specific agent ends."""
        print(f"My agent finished")

    When to use:

    • Agent-specific logging
    • Per-agent monitoring
    • Agent-specific business logic
    • Custom agent behavior

    Run Hooks Methods

    on_agent_start

    Called when an agent starts (any agent in the run):

    Python
    async def on_agent_start(
    self,
    context: AgentHookContext[TContext],
    agent: TAgent,
) -> None:
    """Called before the agent is invoked.
    
    Called each time the current agent changes.
    """
    print(f"Agent {agent.name} starting")

    Use cases:

    • Log agent starts
    • Initialize agent-specific state
    • Track agent usage
    • Validate agent configuration
    on_agent_end

    Called when an agent produces a final output:

    Python
    async def on_agent_end(
    self,
    context: AgentHookContext[TContext],
    agent: TAgent,
    output: Any,
) -> None:
    """Called when the agent produces a final output."""
    print(f"Agent {agent.name} finished with output: {output}")

    Use cases:

    • Log agent outputs
    • Track agent performance
    • Validate output
    • Trigger follow-up actions
    on_handoff

    Called when a handoff occurs:

    Python
    async def on_handoff(
    self,
    context: RunContextWrapper[TContext],
    from_agent: TAgent,
    to_agent: TAgent,
) -> None:
    """Called when a handoff occurs."""
    print(f"Handoff from {from_agent.name} to {to_agent.name}")

    Use cases:

    • Log handoffs
    • Track handoff patterns
    • Validate handoffs
    • Trigger handoff-specific logic
    on_llm_start

    Called just before invoking the LLM:

    Python
    async def on_llm_start(
    self,
    context: RunContextWrapper[TContext],
    agent: Agent[TContext],
    system_prompt: str | None,
    input_items: list[TResponseInputItem],
) -> None:
    """Called just before invoking the LLM for this agent."""
    print(f"LLM call starting for {agent.name}")

    Use cases:

    • Log LLM calls
    • Track prompt usage
    • Validate prompts
    • Modify prompts (with care)
    on_llm_end

    Called immediately after the LLM call returns:

    Python
    async def on_llm_end(
    self,
    context: RunContextWrapper[TContext],
    agent: Agent[TContext],
    response: ModelResponse,
) -> None:
    """Called immediately after the LLM call returns for this agent."""
    print(f"LLM call completed for {agent.name}")

    Use cases:

    • Log LLM responses
    • Track token usage
    • Validate responses
    • Analyze model behavior
    on_tool_start

    Called immediately before a local tool is invoked:

    Python
    async def on_tool_start(
    self,
    context: RunContextWrapper[TContext],
    agent: TAgent,
    tool: Tool,
) -> None:
    """Called immediately before a local tool is invoked.
    
    For function-tool invocations, context is typically a ToolContext instance.
    """
    print(f"Tool {tool.name} starting")

    Use cases:

    • Log tool calls
    • Track tool usage
    • Validate tool inputs
    • Modify tool inputs (with care)
    on_tool_end

    Called immediately after a local tool is invoked:

    Python
    async def on_tool_end(
    self,
    context: RunContextWrapper[TContext],
    agent: TAgent,
    tool: Tool,
    result: str,
) -> None:
    """Called immediately after a local tool is invoked."""
    print(f"Tool {tool.name} finished")

    Use cases:

    • Log tool results
    • Track tool performance
    • Validate tool outputs
    • Analyze tool behavior

    Agent Hooks Methods

    on_start

    Called when this specific agent starts:

    Python
    async def on_start(
    self,
    context: AgentHookContext[TContext],
    agent: TAgent,
) -> None:
    """Called before the agent is invoked.
    
    Called each time the running agent is changed to this agent.
    """
    print(f"This agent starting")

    Use cases:

    • Agent-specific initialization
    • Agent-specific logging
    • Validate agent state
    • Load agent-specific resources
    on_end

    Called when this specific agent produces a final output:

    Python
    async def on_end(
    self,
    context: AgentHookContext[TContext],
    agent: TAgent,
    output: Any,
) -> None:
    """Called when the agent produces a final output."""
    print(f"This agent finished")

    Use cases:

    • Agent-specific cleanup
    • Agent-specific output processing
    • Save agent-specific state
    • Trigger agent-specific follow-ups
    on_handoff

    Called when the agent is being handed off to:

    Python
    async def on_handoff(
    self,
    context: RunContextWrapper[TContext],
    agent: TAgent,
    source: TAgent,
) -> None:
    """Called when the agent is being handed off to.
    
    The source is the agent that is handing off to this agent.
    """
    print(f"Handed off from {source.name} to {agent.name}")

    Use cases:

    • Log handoff reception
    • Prepare agent for handoff
    • Validate handoff context
    • Initialize agent state
    on_tool_start

    Called immediately before a local tool is invoked:

    Python
    async def on_tool_start(
    self,
    context: RunContextWrapper[TContext],
    agent: TAgent,
    tool: Tool,
) -> None:
    """Called immediately before a local tool is invoked."""
    print(f"Tool {tool.name} starting in this agent")

    Use cases:

    • Agent-specific tool logging
    • Track tool usage by agent
    • Agent-specific tool validation
    on_tool_end

    Called immediately after a local tool is invoked:

    Python
    async def on_tool_end(
    self,
    context: RunContextWrapper[TContext],
    agent: TAgent,
    tool: Tool,
    result: str,
) -> None:
    """Called immediately after a local tool is invoked."""
    print(f"Tool {tool.name} finished in this agent")

    Use cases:

    • Agent-specific tool result logging
    • Track tool performance by agent
    • Agent-specific tool result validation
    on_llm_start

    Called immediately before the agent issues an LLM call:

    Python
    async def on_llm_start(
    self,
    context: RunContextWrapper[TContext],
    agent: Agent[TContext],
    system_prompt: str | None,
    input_items: list[TResponseInputItem],
) -> None:
    """Called immediately before the agent issues an LLM call."""
    print(f"LLM call starting for this agent")

    Use cases:

    • Agent-specific LLM logging
    • Track LLM usage by agent
    • Agent-specific prompt validation
    on_llm_end

    Called immediately after the agent receives the LLM response:

    Python
    async def on_llm_end(
    self,
    context: RunContextWrapper[TContext],
    agent: Agent[TContext],
    response: ModelResponse,
) -> None:
    """Called immediately after the agent receives the LLM response."""
    print(f"LLM call completed for this agent")

    Use cases:

    • Agent-specific LLM response logging
    • Track LLM performance by agent
    • Agent-specific response validation

    Using Run Hooks

    Setting Run Hooks

    Set hooks when running an agent:

    Python
    from agents import RunHooks

class MyRunHooks(RunHooks):
    async def on_agent_start(self, context, agent):
        print(f"Agent {agent.name} starting")
    
    async def on_agent_end(self, context, agent, output):
        print(f"Agent {agent.name} finished")

hooks = MyRunHooks()

result = await Runner.run(
    agent,
    input,
    hooks=hooks,
)
    Run Hooks with Context

    Run hooks have access to context:

    Python
    class MyRunHooks(RunHooks[UserContext]):
    async def on_agent_start(self, context, agent):
        user_id = context.context.user_id
        print(f"Agent {agent.name} starting for user {user_id}")
    Run Hooks with RunConfig

    Hooks can be set in RunConfig:

    Python
    config = RunConfig(
    run_hooks=MyRunHooks(),
)

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

    Using Agent Hooks

    Setting Agent Hooks

    Set hooks on an agent:

    Python
    from agents import Agent, AgentHooks

class MyAgentHooks(AgentHooks):
    async def on_start(self, context, agent):
        print(f"My agent starting")
    
    async def on_end(self, context, agent, output):
        print(f"My agent finished")

agent = Agent(
    name="my_agent",
    instructions="...",
    hooks=MyAgentHooks(),
)
    Agent Hooks with Context

    Agent hooks have access to context:

    Python
    class MyAgentHooks(AgentHooks[UserContext]):
    async def on_start(self, context, agent):
        user_id = context.context.user_id
        print(f"My agent starting for user {user_id}")

    Hook Context Types

    AgentHookContext

    Context for agent lifecycle hooks:

    Python
    @dataclass
class AgentHookContext(Generic[TContext]):
    context: TContext
    """User-provided context."""
    
    run_config: RunConfig
    """Run configuration."""
    
    agent: Agent[TContext]
    """The agent."""
    
    session: Session | None
    """Session if configured."""
    RunContextWrapper

    Context for run-level hooks:

    Python
    @dataclass
class RunContextWrapper(Generic[TContext]):
    context: TContext
    """User-provided context."""
    
    usage: Usage
    """Token usage tracking."""
    
    approve_tool: Callable
    """Tool approval function."""
    
    reject_tool: Callable
    """Tool rejection function."""
    
    # ... other fields

    Hook Execution Order

    Complete Hook Order

    Hooks execute in this order:

    Plain text
    1. RunHooks.on_agent_start
2. AgentHooks.on_start (for the starting agent)
3. RunHooks.on_llm_start
4. AgentHooks.on_llm_start
5. [LLM call]
6. AgentHooks.on_llm_end
7. RunHooks.on_llm_end
8. [Tool execution - if tools called]
   - RunHooks.on_tool_start
   - AgentHooks.on_tool_start
   - [Tool execution]
   - AgentHooks.on_tool_end
   - RunHooks.on_tool_end
9. [Handoff - if handoff occurs]
   - RunHooks.on_handoff
   - AgentHooks.on_handoff (for target agent)
10. AgentHooks.on_end (for finishing agent)
11. RunHooks.on_agent_end

    Hook Best Practices

    1. Keep Hooks Fast

    Hooks should be fast to avoid slowing down execution:

    Python
    # Good - fast hook
async def on_agent_start(self, context, agent):
    logger.info(f"Agent {agent.name} starting")  # Fast logging

# Avoid - slow hook
async def on_agent_start(self, context, agent):
    await slow_external_api_call(agent.name)  # Blocks execution
    2. Handle Errors Gracefully

    Hooks should handle errors without breaking execution:

    Python
    # Good - handle errors
async def on_agent_start(self, context, agent):
    try:
        log_to_external_system(agent.name)
    except Exception as e:
        logger.error(f"Hook error: {e}")  # Don't break execution

# Avoid - unhandled errors
async def on_agent_start(self, context, agent):
    log_to_external_system(agent.name)  # Might crash
    3. Use Hooks for Cross-Cutting Concerns

    Use hooks for cross-cutting concerns:

    Python
    # Good - logging is cross-cutting
class LoggingHooks(RunHooks):
    async def on_agent_start(self, context, agent):
        logger.info(f"Agent {agent.name} starting")

# Avoid - business logic in hooks
class BusinessLogicHooks(RunHooks):
    async def on_agent_start(self, context, agent):
        if complex_business_rule(agent):
            do_something()  # Business logic should be in tools/agents
    4. Don't Modify Critical State

    Avoid modifying critical state in hooks:

    Python
    # Good - read-only access
async def on_agent_start(self, context, agent):
    user_id = context.context.user_id
    logger.info(f"User: {user_id}")

# Avoid - modifying state
async def on_agent_start(self, context, agent):
    context.context.user_id = "modified"  # Unexpected behavior
    5. Document Hook Behavior

    Document what hooks do:

    Python
    class MyHooks(RunHooks):
    """Hooks for monitoring agent execution."""
    
    async def on_agent_start(self, context, agent):
        """Log agent start for monitoring."""
        logger.info(f"Agent {agent.name} starting")

    Common Hook Patterns

    1. Logging Hook

    Comprehensive logging:

    Python
    class LoggingHooks(RunHooks):
    """Comprehensive logging hooks."""
    
    async def on_agent_start(self, context, agent):
        logger.info(f"Agent {agent.name} starting")
    
    async def on_agent_end(self, context, agent, output):
        logger.info(f"Agent {agent.name} finished")
    
    async def on_llm_start(self, context, agent, system_prompt, input_items):
        logger.info(f"LLM call starting for {agent.name}")
    
    async def on_llm_end(self, context, agent, response):
        logger.info(f"LLM call completed, tokens: {response.usage.total_tokens}")
    
    async def on_tool_start(self, context, agent, tool):
        logger.info(f"Tool {tool.name} starting")
    
    async def on_tool_end(self, context, agent, tool, result):
        logger.info(f"Tool {tool.name} finished")
    2. Monitoring Hook

    Performance monitoring:

    Python
    class MonitoringHooks(RunHooks):
    """Performance monitoring hooks."""
    
    async def on_llm_end(self, context, agent, response):
        metrics.record(
            "llm_call",
            agent=agent.name,
            tokens=response.usage.total_tokens,
            duration=response.duration,
        )
    
    async def on_tool_end(self, context, agent, tool, result):
        metrics.record(
            "tool_call",
            tool=tool.name,
            duration=result.duration,
        )
    3. Analytics Hook

    Usage analytics:

    Python
    class AnalyticsHooks(RunHooks):
    """Analytics collection hooks."""
    
    async def on_agent_end(self, context, agent, output):
        analytics.track(
            "agent_completion",
            agent=agent.name,
            output_length=len(str(output)),
            turn_count=context.context.turn_count,
        )
    4. Validation Hook

    Custom validation:

    Python
    class ValidationHooks(RunHooks):
    """Validation hooks."""
    
    async def on_llm_end(self, context, agent, response):
        # Validate response
        if not response.response.output:
            raise ValueError("Empty response")
    
    async def on_tool_end(self, context, agent, tool, result):
        # Validate tool result
        if "error" in result.lower():
            logger.warning(f"Tool error: {tool.name}")
    5. State Tracking Hook

    Track state across execution:

    Python
    class StateTrackingHooks(RunHooks):
    """State tracking hooks."""
    
    def __init__(self):
        self.state = {}
    
    async def on_agent_start(self, context, agent):
        self.state[f"{agent.name}_start"] = datetime.now()
    
    async def on_agent_end(self, context, agent, output):
        self.state[f"{agent.name}_end"] = datetime.now()
        duration = (
            self.state[f"{agent.name}_end"] - 
            self.state[f"{agent.name}_start"]
        )
        self.state[f"{agent.name}_duration"] = duration

    Hooks and Tracing

    Hooks Create Trace Spans

    Hooks automatically create trace spans:

    Python
    class MyHooks(RunHooks):
    async def on_agent_start(self, context, agent):
        # This creates a trace span automatically
        print(f"Agent {agent.name} starting")
    Custom Spans in Hooks

    Create custom spans in hooks:

    Python
    from agents import custom_span

class MyHooks(RunHooks):
    async def on_agent_start(self, context, agent):
        with custom_span(name="custom_init"):
            do_custom_initialization()

    Hooks and Context

    Accessing Context in Hooks

    Hooks have access to context:

    Python
    class MyHooks(RunHooks[UserContext]):
    async def on_agent_start(self, context, agent):
        user_id = context.context.user_id
        print(f"User: {user_id}")
    Modifying Context in Hooks

    Context can be modified in hooks (with caution):

    Python
    class MyHooks(RunHooks[UserContext]):
    async def on_agent_start(self, context, agent):
        # Add metadata to context
        context.context.metadata = {
            "start_time": datetime.now(),
            "agent": agent.name,
        }

    Hooks and Errors

    Error Handling in Hooks

    Errors in hooks are logged but don't break execution:

    Python
    class MyHooks(RunHooks):
    async def on_agent_start(self, context, agent):
        try:
            risky_operation()
        except Exception as e:
            logger.error(f"Hook error: {e}")
            # Execution continues
    Hook Error Propagation

    Some errors might propagate:

    Python
    class MyHooks(RunHooks):
    async def on_agent_start(self, context, agent):
        if critical_condition:
            raise CriticalError("Cannot proceed")  # Might break execution

    Hooks and Testing

    Testing Hooks

    Test hooks independently:

    Python
    @pytest.mark.asyncio
async def test_hooks():
    hooks = MyHooks()
    context = create_test_context()
    agent = create_test_agent()
    
    await hooks.on_agent_start(context, agent)
    assert hooks.start_called
    
    await hooks.on_agent_end(context, agent, "output")
    assert hooks.end_called
    Mock Hooks for Testing

    Use mock hooks in tests:

    Python
    class MockHooks(RunHooks):
    def __init__(self):
        self.events = []
    
    async def on_agent_start(self, context, agent):
        self.events.append(("start", agent.name))

mock_hooks = MockHooks()
result = await Runner.run(agent, input, hooks=mock_hooks)
assert ("start", agent.name) in mock_hooks.events

    Hooks Performance

    Hook Overhead

    Hooks add minimal overhead:

    Python
    # Hooks are async and non-blocking
# Fast hooks have negligible impact
# Slow hooks can slow down execution
    Optimizing Hooks

    Optimize hooks for performance:

    Python
    # Option 1: Use fast logging
import logging
logger = logging.getLogger(__name__)  # Fast

# Option 2: Batch operations
class BatchHooks(RunHooks):
    def __init__(self):
        self.batch = []
    
    async def on_agent_end(self, context, agent, output):
        self.batch.append((agent.name, output))
        if len(self.batch) >= 100:
            await flush_batch(self.batch)
            self.batch = []

# Option 3: Async operations
class AsyncHooks(RunHooks):
    async def on_agent_end(self, context, agent, output):
        # Fire and forget
        asyncio.create_task(async_log(agent.name, output))

    Summary

    Lifecycle Hooks enable custom logic at execution points. Key takeaways:

    1. RunHooks apply to the entire run
    2. AgentHooks apply to specific agents
    3. on_agent_start/end track agent lifecycle
    4. on_handoff tracks agent delegation
    5. on_llm_start/end track LLM calls
    6. on_tool_start/end track tool execution
    7. Hook context provides execution information
    8. AgentHookContext for agent-level hooks
    9. RunContextWrapper for run-level hooks
    10. Execution order is predictable
    11. Keep hooks fast to avoid slowdowns
    12. Handle errors gracefully in hooks
    13. Use hooks for cross-cutting concerns
    14. Don't modify critical state in hooks
    15. Document hook behavior clearly
    16. Logging hooks for observability
    17. Monitoring hooks for performance
    18. Analytics hooks for usage tracking
    19. Validation hooks for custom checks
    20. State tracking hooks for execution tracking

    Lifecycle Hooks are essential for extending agent behavior without modifying core code.