feat: Add tool origin tracking to ToolCallItem and ToolCallOutputItem

- Add ToolOriginType enum and ToolOrigin dataclass
- Add _tool_origin field to FunctionTool
- Set tool_origin for MCP tools and agent-as-tool
- Extract and set tool_origin in ToolCallItem and ToolCallOutputItem creation
- Add comprehensive tests for tool origin tracking

Author: habema

src/agents/agent.py

@@ -45,6 +45,8 @@
     FunctionToolResult,
     Tool,
     ToolErrorFunction,
+    ToolOrigin,
+    ToolOriginType,
     _extract_tool_argument_json_error,
     default_tool_error_function,
 )
@@ -802,6 +804,11 @@ async def _run_agent_tool(context: ToolContext, input_json: str) -> Any:
         )
         run_agent_tool._is_agent_tool = True
         run_agent_tool._agent_instance = self
+        # Set origin tracking on run_agent (the FunctionTool returned by @function_tool)
+        run_agent_tool._tool_origin = ToolOrigin(
+            type=ToolOriginType.AGENT_AS_TOOL,
+            agent_as_tool=self,
+        )
 
         return run_agent_tool
 

src/agents/items.py

@@ -49,6 +49,7 @@
 from .exceptions import AgentsException, ModelBehaviorError
 from .logger import logger
 from .tool import (
+    ToolOrigin,
     ToolOutputFileContent,
     ToolOutputImage,
     ToolOutputText,
@@ -248,6 +249,9 @@ class ToolCallItem(RunItemBase[Any]):
     description: str | None = None
     """Optional tool description if known at item creation time."""
 
+    tool_origin: ToolOrigin | None = field(default=None, repr=False)
+    """Information about the origin/source of the tool call. Only set for FunctionTool calls."""
+
 
 ToolCallOutputTypes: TypeAlias = Union[
     FunctionCallOutput,
@@ -271,6 +275,9 @@ class ToolCallOutputItem(RunItemBase[Any]):
 
     type: Literal["tool_call_output_item"] = "tool_call_output_item"
 
+    tool_origin: ToolOrigin | None = field(default=None, repr=False)
+    """Information about the origin/source of the tool call. Only set for FunctionTool calls."""
+
     def to_input_item(self) -> TResponseInputItem:
         """Converts the tool output into an input item for the next model turn.
 

src/agents/mcp/util.py

@@ -20,6 +20,8 @@
     FunctionTool,
     Tool,
     ToolErrorFunction,
+    ToolOrigin,
+    ToolOriginType,
     ToolOutputImageDict,
     ToolOutputTextDict,
     default_tool_error_function,
@@ -301,14 +303,19 @@ async def invoke_func(ctx: ToolContext[Any], input_json: str) -> ToolOutput:
             bool | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]]
         ) = server._get_needs_approval_for_tool(tool, agent)
 
-        return FunctionTool(
+        function_tool = FunctionTool(
             name=tool.name,
             description=tool.description or "",
             params_json_schema=schema,
             on_invoke_tool=invoke_func,
             strict_json_schema=is_strict,
             needs_approval=needs_approval,
         )
+        function_tool._tool_origin = ToolOrigin(
+            type=ToolOriginType.MCP,
+            mcp_server=server,
+        )
+        return function_tool
 
     @staticmethod
     def _merge_mcp_meta(

src/agents/run_internal/run_loop.py

@@ -49,7 +49,7 @@
     RawResponsesStreamEvent,
     RunItemStreamEvent,
 )
-from ..tool import Tool, dispose_resolved_computers
+from ..tool import FunctionTool, Tool, _get_tool_origin_info, dispose_resolved_computers
 from ..tracing import Span, SpanError, agent_span, get_current_trace
 from ..tracing.model_tracing import get_model_tracing_impl
 from ..tracing.span_data import AgentSpanData
@@ -1216,13 +1216,18 @@ async def run_single_turn_streamed(
                     # execution behavior in process_model_response).
                     tool_name = getattr(output_item, "name", None)
                     tool_description: str | None = None
+                    tool_origin = None
                     if isinstance(tool_name, str) and tool_name in tool_map:
-                        tool_description = getattr(tool_map[tool_name], "description", None)
+                        tool = tool_map[tool_name]
+                        tool_description = getattr(tool, "description", None)
+                        if isinstance(tool, FunctionTool):
+                            tool_origin = _get_tool_origin_info(tool)
 
                     tool_item = ToolCallItem(
                         raw_item=cast(ToolCallItemTypes, output_item),
                         agent=agent,
                         description=tool_description,
+                        tool_origin=tool_origin,
                     )
                     streamed_result._event_queue.put_nowait(
                         RunItemStreamEvent(item=tool_item, name="tool_called")

src/agents/run_internal/tool_execution.py

@@ -52,6 +52,7 @@
     ShellCallOutcome,
     ShellCommandOutput,
     Tool,
+    _get_tool_origin_info,
     resolve_computer,
 )
 from ..tool_context import ToolContext
@@ -973,10 +974,12 @@ async def run_single_tool(func_tool: FunctionTool, tool_call: ResponseFunctionTo
 
             run_item: RunItem | None = None
             if not nested_interruptions:
+                tool_origin = _get_tool_origin_info(tool_run.function_tool)
                 run_item = ToolCallOutputItem(
                     output=result,
                     raw_item=ItemHelpers.tool_call_output_item(tool_run.tool_call, result),
                     agent=agent,
+                    tool_origin=tool_origin,
                 )
             else:
                 # Skip tool output until nested interruptions are resolved.

src/agents/run_internal/turn_resolution.py

@@ -62,6 +62,7 @@
     LocalShellTool,
     ShellTool,
     Tool,
+    _get_tool_origin_info,
 )
 from ..tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult
 from ..tracing import SpanError, handoff_span
@@ -1473,8 +1474,14 @@ def process_model_response(
                 raise ModelBehaviorError(error)
 
             func_tool = function_map[output.name]
+            tool_origin = _get_tool_origin_info(func_tool)
             items.append(
-                ToolCallItem(raw_item=output, agent=agent, description=func_tool.description)
+                ToolCallItem(
+                    raw_item=output,
+                    agent=agent,
+                    description=func_tool.description,
+                    tool_origin=tool_origin,
+                )
             )
             functions.append(
                 ToolRunFunction(

src/agents/tool.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import enum
 import inspect
 import json
 import weakref
@@ -48,6 +49,7 @@
 if TYPE_CHECKING:
     from .agent import Agent, AgentBase
     from .items import RunItem, ToolApprovalItem
+    from .mcp.server import MCPServer
 
 
 ToolParams = ParamSpec("ToolParams")
@@ -182,6 +184,59 @@ class ComputerProvider(Generic[ComputerT]):
 ]
 
 
+class ToolOriginType(str, enum.Enum):
+    """The type of tool origin."""
+
+    FUNCTION = "function"
+    """Regular Python function tool created via @function_tool decorator."""
+
+    MCP = "mcp"
+    """MCP server tool converted via MCPUtil.to_function_tool()."""
+
+    AGENT_AS_TOOL = "agent_as_tool"
+    """Agent converted to tool via agent.as_tool()."""
+
+
+@dataclass
+class ToolOrigin:
+    """Information about the origin/source of a function tool."""
+
+    type: ToolOriginType
+    """The type of tool origin."""
+
+    mcp_server: MCPServer | None = None
+    """The MCP server object. Only set when type is MCP."""
+
+    agent_as_tool: Agent[Any] | None = None
+    """The agent object. Only set when type is AGENT_AS_TOOL."""
+
+    def __repr__(self) -> str:
+        """Custom repr that only includes relevant fields."""
+        parts = [f"type={self.type.value!r}"]
+        if self.mcp_server is not None:
+            parts.append(f"mcp_server_name={self.mcp_server.name!r}")
+        if self.agent_as_tool is not None:
+            parts.append(f"agent_as_tool_name={self.agent_as_tool.name!r}")
+        return f"ToolOrigin({', '.join(parts)})"
+
+
+def _get_tool_origin_info(function_tool: FunctionTool) -> ToolOrigin | None:
+    """Extract origin information from a FunctionTool.
+
+    Args:
+        function_tool: The function tool to extract origin info from.
+
+    Returns:
+        ToolOrigin object if origin is set, otherwise None (defaults to FUNCTION type).
+    """
+    origin = function_tool._tool_origin
+    if origin is None:
+        # Default to FUNCTION if not explicitly set
+        return ToolOrigin(type=ToolOriginType.FUNCTION)
+
+    return origin
+
+
 @dataclass
 class FunctionToolResult:
     tool: FunctionTool
@@ -264,6 +319,9 @@ class FunctionTool:
     _agent_instance: Any = field(default=None, init=False, repr=False)
     """Internal reference to the agent instance if this is an agent-as-tool."""
 
+    _tool_origin: ToolOrigin | None = field(default=None, init=False, repr=False)
+    """Internal field tracking the origin of this tool (FUNCTION, MCP, or AGENT_AS_TOOL)."""
+
     def __post_init__(self):
         if self.strict_json_schema:
             self.params_json_schema = ensure_strict_json_schema(self.params_json_schema)