feat: support agent output guardrails in realtime (#1381)

## Summary
- allow RealtimeAgent to define output guardrails
- run agent-level guardrails in realtime sessions
- avoid running duplicate guardrails when supplied on both agent and run
config
- document and test agent-level output guardrails

## Testing
- `make format`
- `make lint`
- `make mypy`
- `make tests`
- `make build-docs`


------
https://chatgpt.com/codex/tasks/task_i_6892a5f23df0832492b8cba9ed214272

Author: jhills20

docs/ja/realtime/guide.md

@@ -134,6 +134,24 @@ main_agent = RealtimeAgent(
 
 Realtime エージェントでは出力ガードレールのみがサポートされます。パフォーマンス低下を防ぐためにデバウンスされ、リアルタイム生成中に毎単語ではなく定期的に実行されます。デフォルトのデバウンス長は 100 文字で、設定可能です。
 
+ガードレールは `RealtimeAgent` に直接設定することも、セッションの `run_config` で指定することもできます。両方のリストが同時に実行されます。
+
+```python
+from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail
+
+def sensitive_data_check(context, agent, output):
+    return GuardrailFunctionOutput(
+        tripwire_triggered="password" in output,
+        output_info=None,
+    )
+
+agent = RealtimeAgent(
+    name="Assistant",
+    instructions="...",
+    output_guardrails=[OutputGuardrail(guardrail_function=sensitive_data_check)],
+)
+```
+
 ガードレールがトリガーされると `guardrail_tripped` イベントが生成され、エージェントの現在の応答を中断できます。デバウンス動作により安全性とリアルタイム性能のバランスを取ります。テキストエージェントとは異なり、realtime エージェントはガードレールがトリップしても Exception を発生させません。
 
 ## オーディオ処理

docs/realtime/guide.md

@@ -130,6 +130,24 @@ For complete event details, see [`RealtimeSessionEvent`][agents.realtime.events.
 
 Only output guardrails are supported for realtime agents. These guardrails are debounced and run periodically (not on every word) to avoid performance issues during real-time generation. The default debounce length is 100 characters, but this is configurable.
 
+Guardrails can be attached directly to a `RealtimeAgent` or provided via the session's `run_config`. Guardrails from both sources run together.
+
+```python
+from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail
+
+def sensitive_data_check(context, agent, output):
+    return GuardrailFunctionOutput(
+        tripwire_triggered="password" in output,
+        output_info=None,
+    )
+
+agent = RealtimeAgent(
+    name="Assistant",
+    instructions="...",
+    output_guardrails=[OutputGuardrail(guardrail_function=sensitive_data_check)],
+)
+```
+
 When a guardrail is triggered, it generates a `guardrail_tripped` event and can interrupt the agent's current response. The debounce behavior helps balance safety with real-time performance requirements. Unlike text agents, realtime agents do **not** raise an Exception when guardrails are tripped.
 
 ## Audio processing

src/agents/realtime/agent.py

@@ -7,6 +7,7 @@
 from typing import Any, Callable, Generic, cast
 
 from ..agent import AgentBase
+from ..guardrail import OutputGuardrail
 from ..handoffs import Handoff
 from ..lifecycle import AgentHooksBase, RunHooksBase
 from ..logger import logger
@@ -62,6 +63,11 @@ class RealtimeAgent(AgentBase, Generic[TContext]):
     modularity.
     """
 
+    output_guardrails: list[OutputGuardrail[TContext]] = field(default_factory=list)
+    """A list of checks that run on the final output of the agent, after generating a response.
+    Runs only if the agent produces a final output.
+    """
+
     hooks: RealtimeAgentHooks | None = None
     """A class that receives callbacks on various lifecycle events for this agent.
     """

src/agents/realtime/session.py

@@ -444,7 +444,17 @@ def _get_new_history(
 
     async def _run_output_guardrails(self, text: str) -> bool:
         """Run output guardrails on the given text. Returns True if any guardrail was triggered."""
-        output_guardrails = self._run_config.get("output_guardrails", [])
+        combined_guardrails = self._current_agent.output_guardrails + self._run_config.get(
+            "output_guardrails", []
+        )
+        seen_ids: set[int] = set()
+        output_guardrails = []
+        for guardrail in combined_guardrails:
+            guardrail_id = id(guardrail)
+            if guardrail_id not in seen_ids:
+                output_guardrails.append(guardrail)
+                seen_ids.add(guardrail_id)
+
         if not output_guardrails or self._interrupted_by_guardrail:
             return False
 

tests/realtime/test_session.py

@@ -104,6 +104,7 @@ def mock_agent():
     agent.get_all_tools = AsyncMock(return_value=[])
 
     type(agent).handoffs = PropertyMock(return_value=[])
+    type(agent).output_guardrails = PropertyMock(return_value=[])
     return agent
 
 
@@ -1063,6 +1064,37 @@ async def test_transcript_delta_triggers_guardrail_at_threshold(
         assert len(guardrail_events) == 1
         assert guardrail_events[0].message == "this is more than ten characters"
 
+    @pytest.mark.asyncio
+    async def test_agent_and_run_config_guardrails_not_run_twice(self, mock_model):
+        """Guardrails shared by agent and run config should execute once."""
+
+        call_count = 0
+
+        def guardrail_func(context, agent, output):
+            nonlocal call_count
+            call_count += 1
+            return GuardrailFunctionOutput(output_info={}, tripwire_triggered=False)
+
+        shared_guardrail = OutputGuardrail(
+            guardrail_function=guardrail_func, name="shared_guardrail"
+        )
+
+        agent = RealtimeAgent(name="agent", output_guardrails=[shared_guardrail])
+        run_config: RealtimeRunConfig = {
+            "output_guardrails": [shared_guardrail],
+            "guardrails_settings": {"debounce_text_length": 5},
+        }
+
+        session = RealtimeSession(mock_model, agent, None, run_config=run_config)
+
+        await session.on_event(
+            RealtimeModelTranscriptDeltaEvent(item_id="item_1", delta="hello", response_id="resp_1")
+        )
+
+        await self._wait_for_guardrail_tasks(session)
+
+        assert call_count == 1
+
     @pytest.mark.asyncio
     async def test_transcript_delta_multiple_thresholds_same_item(
         self, mock_model, mock_agent, triggered_guardrail
@@ -1210,6 +1242,36 @@ def guardrail_func(context, agent, output):
         assert len(guardrail_events) == 1
         assert len(guardrail_events[0].guardrail_results) == 2
 
+    @pytest.mark.asyncio
+    async def test_agent_output_guardrails_triggered(self, mock_model, triggered_guardrail):
+        """Test that guardrails defined on the agent are executed."""
+        agent = RealtimeAgent(name="agent", output_guardrails=[triggered_guardrail])
+        run_config: RealtimeRunConfig = {
+            "guardrails_settings": {"debounce_text_length": 10},
+        }
+
+        session = RealtimeSession(mock_model, agent, None, run_config=run_config)
+
+        transcript_event = RealtimeModelTranscriptDeltaEvent(
+            item_id="item_1", delta="this is more than ten characters", response_id="resp_1"
+        )
+
+        await session.on_event(transcript_event)
+        await self._wait_for_guardrail_tasks(session)
+
+        assert session._interrupted_by_guardrail is True
+        assert mock_model.interrupts_called == 1
+        assert len(mock_model.sent_messages) == 1
+        assert "triggered_guardrail" in mock_model.sent_messages[0]
+
+        events = []
+        while not session._event_queue.empty():
+            events.append(await session._event_queue.get())
+
+        guardrail_events = [e for e in events if isinstance(e, RealtimeGuardrailTripped)]
+        assert len(guardrail_events) == 1
+        assert guardrail_events[0].message == "this is more than ten characters"
+
 
 class TestModelSettingsIntegration:
     """Test suite for model settings integration in RealtimeSession."""