feat: expose immutable agent tool invocation metadata on run results (ref #2575) (#2576)

Author: seratch

src/agents/__init__.py

@@ -86,7 +86,7 @@
 from .prompts import DynamicPromptFunction, GenerateDynamicPromptData, Prompt
 from .repl import run_demo_loop
 from .responses_websocket_session import ResponsesWebSocketSession, responses_websocket_session
-from .result import RunResult, RunResultStreaming
+from .result import AgentToolInvocation, RunResult, RunResultStreaming
 from .run import (
     ReasoningItemIdPolicy,
     RunConfig,
@@ -359,6 +359,7 @@ def enable_verbose_stdout_logging():
     "RunErrorHandlerInput",
     "RunErrorHandlerResult",
     "RunErrorHandlers",
+    "AgentToolInvocation",
     "RunResult",
     "RunResultStreaming",
     "ResponsesWebSocketSession",

src/agents/agent.py

@@ -501,7 +501,8 @@ def as_tool(
             tool_description: The description of the tool, which should indicate what it does and
                 when to use it.
             custom_output_extractor: A function that extracts the output from the agent. If not
-                provided, the last message from the agent will be used.
+                provided, the last message from the agent will be used. Nested run results expose
+                `agent_tool_invocation` metadata when this agent is invoked via `as_tool()`.
             is_enabled: Whether the tool is enabled. Can be a bool or a callable that takes the run
                 context and agent and returns whether the tool is enabled. Disabled tools are hidden
                 from the LLM at runtime.

src/agents/result.py

@@ -46,11 +46,25 @@
 )
 
 if TYPE_CHECKING:
-    from .tool_context import ToolContext
+    pass
 
 T = TypeVar("T")
 
 
+@dataclass(frozen=True)
+class AgentToolInvocation:
+    """Immutable metadata about a nested agent-tool invocation."""
+
+    tool_name: str
+    """The nested tool name exposed to the model."""
+
+    tool_call_id: str
+    """The tool call ID for the nested invocation."""
+
+    tool_arguments: str
+    """The raw JSON arguments for the nested invocation."""
+
+
 def _populate_state_from_result(
     state: RunState[Any],
     result: RunResultBase,
@@ -208,13 +222,21 @@ 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."""
+    def agent_tool_invocation(self) -> AgentToolInvocation | None:
+        """Immutable metadata for results produced by `Agent.as_tool()`.
+
+        Returns `None` for ordinary top-level runs.
+        """
         from .tool_context import ToolContext
 
-        if isinstance(self.context_wrapper, ToolContext):
-            return self.context_wrapper
-        return None
+        if not isinstance(self.context_wrapper, ToolContext):
+            return None
+
+        return AgentToolInvocation(
+            tool_name=self.context_wrapper.tool_name,
+            tool_call_id=self.context_wrapper.tool_call_id,
+            tool_arguments=self.context_wrapper.tool_arguments,
+        )
 
     @property
     def last_response_id(self) -> str | None:

tests/test_agent_as_tool.py

@@ -395,12 +395,11 @@ async def extractor(result) -> str:
 
 
 @pytest.mark.asyncio
-async def test_agent_as_tool_extractor_can_access_tool_context(
+async def test_agent_as_tool_extractor_can_access_agent_tool_invocation(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     agent = Agent(name="nested_agent")
-
-    real_run_result = RunResult(
+    run_result = RunResult(
         input="hello",
         new_items=[],
         raw_responses=[],
@@ -413,7 +412,7 @@ async def test_agent_as_tool_extractor_can_access_tool_context(
             context=None,
             tool_name="nested_tool",
             tool_call_id="call_abc_123",
-            tool_arguments="{}",
+            tool_arguments='{"input": "hello"}',
         ),
         _last_agent=agent,
     )
@@ -433,16 +432,19 @@ async def fake_run(
     ):
         del cls, starting_agent, input, context, max_turns, hooks, run_config
         del previous_response_id, conversation_id, session
-        return real_run_result
+        return run_result
 
     monkeypatch.setattr(Runner, "run", classmethod(fake_run))
 
-    received_call_id: str | None = None
+    received_tool_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
+        nonlocal received_tool_call_id
+        invocation = result.agent_tool_invocation
+        assert invocation is not None
+        received_tool_call_id = invocation.tool_call_id
+        assert invocation.tool_name == "nested_tool"
+        assert invocation.tool_arguments == '{"input": "hello"}'
         return "extracted"
 
     tool = agent.as_tool(
@@ -460,7 +462,7 @@ async def extractor(result: RunResult | RunResultStreaming) -> str:
     output = await tool.on_invoke_tool(parent_tool_context, '{"input": "hello"}')
 
     assert output == "extracted"
-    assert received_call_id == "call_abc_123"
+    assert received_tool_call_id == "call_abc_123"
 
 
 @pytest.mark.asyncio
@@ -1412,7 +1414,7 @@ async def on_stream(payload: AgentToolStreamEvent) -> None:
 
 
 @pytest.mark.asyncio
-async def test_agent_as_tool_streaming_extractor_can_access_tool_context(
+async def test_agent_as_tool_streaming_extractor_can_access_agent_tool_invocation(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     agent = Agent(name="streaming_tool_context_agent")
@@ -1441,7 +1443,13 @@ async def test_agent_as_tool_streaming_extractor_can_access_tool_context(
     streamed_instance._event_queue.put_nowait(stream_event)
     streamed_instance.is_complete = True
 
-    def fake_run_streamed(cls, /, starting_agent, input, **kwargs) -> RunResultStreaming:
+    def fake_run_streamed(
+        cls,
+        /,
+        starting_agent,
+        input,
+        **kwargs,
+    ) -> RunResultStreaming:
         del cls, starting_agent, input, kwargs
         return streamed_instance
 
@@ -1455,8 +1463,11 @@ async def unexpected_run(*args: Any, **kwargs: Any) -> 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
+        invocation = result.agent_tool_invocation
+        assert invocation is not None
+        received_call_id = invocation.tool_call_id
+        assert invocation.tool_name == "stream_tool"
+        assert invocation.tool_arguments == '{"input": "go"}'
         return "custom value"
 
     async def on_stream(payload: AgentToolStreamEvent) -> None:

tests/test_result_cast.py

@@ -3,21 +3,23 @@
 import dataclasses
 import gc
 import weakref
-from typing import Any
+from typing import Any, cast
 
 import pytest
 from openai.types.responses import ResponseOutputMessage, ResponseOutputText
 from pydantic import BaseModel, ConfigDict
 
 from agents import (
     Agent,
+    AgentToolInvocation,
     MessageOutputItem,
     RunContextWrapper,
     RunItem,
     RunResult,
     RunResultStreaming,
 )
 from agents.exceptions import AgentsException
+from agents.tool_context import ToolContext
 
 
 def create_run_result(
@@ -259,9 +261,13 @@ def test_run_result_streaming_release_agents_releases_current_agent() -> None:
         _ = streaming_result.last_agent
 
 
-def test_run_result_tool_context_returns_tool_context() -> None:
-    from agents.tool_context import ToolContext
+def test_run_result_agent_tool_invocation_returns_none_for_plain_context() -> None:
+    result = create_run_result("ok")
+
+    assert result.agent_tool_invocation is None
 
+
+def test_run_result_agent_tool_invocation_returns_immutable_metadata() -> None:
     tool_ctx = ToolContext(
         context=None,
         tool_name="my_tool",
@@ -282,20 +288,19 @@ def test_run_result_tool_context_returns_tool_context() -> None:
         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
+    assert result.agent_tool_invocation == AgentToolInvocation(
+        tool_name="my_tool",
+        tool_call_id="call_xyz",
+        tool_arguments="{}",
+    )
 
+    invocation = result.agent_tool_invocation
+    assert invocation is not None
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        cast(Any, invocation).tool_name = "other"
 
-def test_run_result_streaming_tool_context_returns_tool_context() -> None:
-    from agents.tool_context import ToolContext
 
+def test_run_result_streaming_agent_tool_invocation_returns_metadata() -> None:
     agent = Agent(name="streaming-tool-agent")
     tool_ctx = ToolContext(
         context=None,
@@ -321,6 +326,8 @@ def test_run_result_streaming_tool_context_returns_tool_context() -> 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"
+    assert result.agent_tool_invocation == AgentToolInvocation(
+        tool_name="stream_tool",
+        tool_call_id="call_stream",
+        tool_arguments='{"input":"stream"}',
+    )