Part 7: Streaming with guardrails

Guardrails also work with streaming responses. A chat UI often shows tokens as they arrive, but unsafe partial output should not be shown if a guardrail is going to block the response.

This section is short because the code is a small change from Runner.run to Runner.run_streamed. It still matters because streaming is where guardrail behavior becomes visible to users.

Start a streamed run

Import the response delta event type:

from openai.types.responses import ResponseTextDeltaEvent

Start the guarded agent in streaming mode:

result = Runner.run_streamed(
    fully_guarded_agent,
    input="Tell me about the certificate"
)

This uses the same fully_guarded_agent from the output-guardrail step. The difference is that we consume events instead of waiting for one final string.

Print text deltas

Loop over stream events and print text deltas:

async for event in result.stream_events():
    if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
        print(event.data.delta, end="")

This is the notebook version of what a web UI would do with streamed tokens. A production UI would append the delta to a message bubble instead of printing to standard output.

Wait for the final result

After consuming the stream, wait for completion and return the final output:

await result.run()
print(result.final_output)

The final output is still useful because it gives you the complete message after all streaming events have been processed.

Add streaming guardrail handling

Wrap the streamed run in the same exception handling style as run_guarded:

async def run_streaming_guarded(agent, user_input):
    """Run streaming agent with guardrail handling."""
    try:
        result = Runner.run_streamed(agent, input=user_input)

        async for event in result.stream_events():
            if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
                print(event.data.delta, end="")

        await result.run()
        return result.final_output

    except InputGuardrailTripwireTriggered as e:
        return f"\n[INPUT BLOCKED] {e.guardrail_result.output.output_info}"
    except OutputGuardrailTripwireTriggered as e:
        return f"\n[OUTPUT BLOCKED] {e.guardrail_result.output.output_info}"

What you want is the exception before unsafe output is shown to the user. Verify that behavior in the framework version you deploy, because streaming changes the user experience.

Continue with Part 8: Tool-based guardrails to build a guardrail pattern that works even when the framework has no guardrail API.