feat: #2247 add RunResult tool_context accessor for agent tools (#2575)

This pull request adds a typed `RunResult.tool_context` convenience
accessor for nested agent-tool runs so `custom_output_extractor`
implementations can inspect `ToolContext` metadata without a cast. The
change keeps the existing `RunResult` and `RunResultStreaming`
constructor shapes intact by adding the accessor on the shared
`RunResultBase`, which keeps the release risk low while improving mypy
ergonomics for agent-as-tool integrations.

It also updates the agent-as-tool tests to cover both non-streaming and
streaming extractor access to `tool_context`, and documents the new
extractor pattern in `docs/tools.md`. Behavior is unchanged for normal
runs: `tool_context` is `None` unless the run was started through
`Agent.as_tool(...)`.

Resolves #2247

Author: seratch

src/agents/result.py

@@ -6,7 +6,7 @@
 import weakref
 from collections.abc import AsyncIterator
 from dataclasses import InitVar, dataclass, field
-from typing import Any, Literal, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast
 
 from pydantic import GetCoreSchemaHandler
 from pydantic_core import core_schema
@@ -45,6 +45,9 @@
     pretty_print_run_result_streaming,
 )
 
+if TYPE_CHECKING:
+    from .tool_context import ToolContext
+
 T = TypeVar("T")
 
 
@@ -204,6 +207,15 @@ def to_input_list(self) -> list[TResponseInputItem]:
 
         return original_items + new_items
 
+    @property
+    def tool_context(self) -> ToolContext[Any] | None:
+        """The tool context for runs started via ``Agent.as_tool()``, if available."""
+        from .tool_context import ToolContext
+
+        if isinstance(self.context_wrapper, ToolContext):
+            return self.context_wrapper
+        return None
+
     @property
     def last_response_id(self) -> str | None:
         """Convenience method to get the response ID of the last model response."""

tests/test_agent_as_tool.py

@@ -19,6 +19,8 @@
     RunContextWrapper,
     RunHooks,
     Runner,
+    RunResult,
+    RunResultStreaming,
     Session,
     SessionSettings,
     ToolApprovalItem,
@@ -392,6 +394,75 @@ async def extractor(result) -> str:
     assert output == "custom output"
 
 
+@pytest.mark.asyncio
+async def test_agent_as_tool_extractor_can_access_tool_context(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    agent = Agent(name="nested_agent")
+
+    real_run_result = RunResult(
+        input="hello",
+        new_items=[],
+        raw_responses=[],
+        final_output="done",
+        input_guardrail_results=[],
+        output_guardrail_results=[],
+        tool_input_guardrail_results=[],
+        tool_output_guardrail_results=[],
+        context_wrapper=ToolContext(
+            context=None,
+            tool_name="nested_tool",
+            tool_call_id="call_abc_123",
+            tool_arguments="{}",
+        ),
+        _last_agent=agent,
+    )
+
+    async def fake_run(
+        cls,
+        starting_agent,
+        input,
+        *,
+        context,
+        max_turns,
+        hooks,
+        run_config,
+        previous_response_id,
+        conversation_id,
+        session,
+    ):
+        del cls, starting_agent, input, context, max_turns, hooks, run_config
+        del previous_response_id, conversation_id, session
+        return real_run_result
+
+    monkeypatch.setattr(Runner, "run", classmethod(fake_run))
+
+    received_call_id: str | None = None
+
+    async def extractor(result: RunResult | RunResultStreaming) -> str:
+        nonlocal received_call_id
+        assert result.tool_context is not None
+        received_call_id = result.tool_context.tool_call_id
+        return "extracted"
+
+    tool = agent.as_tool(
+        tool_name="nested_tool",
+        tool_description="A nested agent tool",
+        custom_output_extractor=extractor,
+    )
+
+    parent_tool_context = ToolContext(
+        context=None,
+        tool_name="nested_tool",
+        tool_call_id="call_abc_123",
+        tool_arguments='{"input": "hello"}',
+    )
+    output = await tool.on_invoke_tool(parent_tool_context, '{"input": "hello"}')
+
+    assert output == "extracted"
+    assert received_call_id == "call_abc_123"
+
+
 @pytest.mark.asyncio
 async def test_agent_as_tool_inherits_parent_run_config_when_not_set(
     monkeypatch: pytest.MonkeyPatch,
@@ -1254,18 +1325,29 @@ async def test_agent_as_tool_streaming_works_with_custom_extractor(
 ) -> None:
     agent = Agent(name="streamer")
     stream_events = [RawResponsesStreamEvent(data=cast(Any, {"type": "response_started"}))]
-    stream_events = [RawResponsesStreamEvent(data=cast(Any, {"type": "response_started"}))]
-
-    class DummyStreamingResult:
-        def __init__(self) -> None:
-            self.final_output = "raw output"
-            self.current_agent = agent
-
-        async def stream_events(self):
-            for ev in stream_events:
-                yield ev
-
-    streamed_instance = DummyStreamingResult()
+    streamed_instance = RunResultStreaming(
+        input="stream please",
+        new_items=[],
+        raw_responses=[],
+        final_output="raw output",
+        input_guardrail_results=[],
+        output_guardrail_results=[],
+        tool_input_guardrail_results=[],
+        tool_output_guardrail_results=[],
+        context_wrapper=ToolContext(
+            context=None,
+            tool_name="stream_tool",
+            tool_call_id="call-abc",
+            tool_arguments='{"input": "stream please"}',
+        ),
+        current_agent=agent,
+        current_turn=0,
+        max_turns=1,
+        _current_agent_output_schema=None,
+        trace=None,
+    )
+    streamed_instance._event_queue.put_nowait(stream_events[0])
+    streamed_instance.is_complete = True
 
     def fake_run_streamed(
         cls,
@@ -1329,6 +1411,76 @@ async def on_stream(payload: AgentToolStreamEvent) -> None:
     assert callbacks == stream_events
 
 
+@pytest.mark.asyncio
+async def test_agent_as_tool_streaming_extractor_can_access_tool_context(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    agent = Agent(name="streaming_tool_context_agent")
+    stream_event = RawResponsesStreamEvent(data=cast(Any, {"type": "response_started"}))
+    streamed_instance = RunResultStreaming(
+        input="go",
+        new_items=[],
+        raw_responses=[],
+        final_output="raw output",
+        input_guardrail_results=[],
+        output_guardrail_results=[],
+        tool_input_guardrail_results=[],
+        tool_output_guardrail_results=[],
+        context_wrapper=ToolContext(
+            context=None,
+            tool_name="stream_tool",
+            tool_call_id="call-stream-123",
+            tool_arguments='{"input": "go"}',
+        ),
+        current_agent=agent,
+        current_turn=0,
+        max_turns=1,
+        _current_agent_output_schema=None,
+        trace=None,
+    )
+    streamed_instance._event_queue.put_nowait(stream_event)
+    streamed_instance.is_complete = True
+
+    def fake_run_streamed(cls, /, starting_agent, input, **kwargs) -> RunResultStreaming:
+        del cls, starting_agent, input, kwargs
+        return streamed_instance
+
+    async def unexpected_run(*args: Any, **kwargs: Any) -> None:
+        raise AssertionError("Runner.run should not be called when on_stream is provided.")
+
+    monkeypatch.setattr(Runner, "run_streamed", classmethod(fake_run_streamed))
+    monkeypatch.setattr(Runner, "run", classmethod(unexpected_run))
+
+    received_call_id: str | None = None
+
+    async def extractor(result: RunResult | RunResultStreaming) -> str:
+        nonlocal received_call_id
+        assert result.tool_context is not None
+        received_call_id = result.tool_context.tool_call_id
+        return "custom value"
+
+    async def on_stream(payload: AgentToolStreamEvent) -> None:
+        del payload
+
+    tool = agent.as_tool(
+        tool_name="stream_tool",
+        tool_description="Streams events",
+        custom_output_extractor=extractor,
+        on_stream=on_stream,
+    )
+
+    tool_context = ToolContext(
+        context=None,
+        tool_name="stream_tool",
+        tool_call_id="call-stream-123",
+        tool_arguments='{"input": "go"}',
+    )
+    output = await tool.on_invoke_tool(tool_context, '{"input": "go"}')
+
+    assert output == "custom value"
+    assert received_call_id == "call-stream-123"
+
+
 @pytest.mark.asyncio
 async def test_agent_as_tool_streaming_accepts_sync_handler(
     monkeypatch: pytest.MonkeyPatch,

tests/test_result_cast.py

@@ -257,3 +257,70 @@ def test_run_result_streaming_release_agents_releases_current_agent() -> None:
     assert agent_ref() is None
     with pytest.raises(AgentsException):
         _ = streaming_result.last_agent
+
+
+def test_run_result_tool_context_returns_tool_context() -> None:
+    from agents.tool_context import ToolContext
+
+    tool_ctx = ToolContext(
+        context=None,
+        tool_name="my_tool",
+        tool_call_id="call_xyz",
+        tool_arguments="{}",
+    )
+    result = RunResult(
+        input="test",
+        new_items=[],
+        raw_responses=[],
+        final_output="ok",
+        input_guardrail_results=[],
+        output_guardrail_results=[],
+        tool_input_guardrail_results=[],
+        tool_output_guardrail_results=[],
+        _last_agent=Agent(name="test"),
+        context_wrapper=tool_ctx,
+        interruptions=[],
+    )
+
+    assert result.tool_context is tool_ctx
+    assert result.tool_context is not None
+    assert result.tool_context.tool_call_id == "call_xyz"
+
+
+def test_run_result_tool_context_returns_none_for_plain_context() -> None:
+    result = create_run_result("ok")
+
+    assert result.tool_context is None
+
+
+def test_run_result_streaming_tool_context_returns_tool_context() -> None:
+    from agents.tool_context import ToolContext
+
+    agent = Agent(name="streaming-tool-agent")
+    tool_ctx = ToolContext(
+        context=None,
+        tool_name="stream_tool",
+        tool_call_id="call_stream",
+        tool_arguments='{"input":"stream"}',
+    )
+    result = RunResultStreaming(
+        input="stream",
+        new_items=[],
+        raw_responses=[],
+        final_output="done",
+        input_guardrail_results=[],
+        output_guardrail_results=[],
+        tool_input_guardrail_results=[],
+        tool_output_guardrail_results=[],
+        context_wrapper=tool_ctx,
+        current_agent=agent,
+        current_turn=0,
+        max_turns=1,
+        _current_agent_output_schema=None,
+        trace=None,
+        interruptions=[],
+    )
+
+    assert result.tool_context is tool_ctx
+    assert result.tool_context is not None
+    assert result.tool_context.tool_call_id == "call_stream"