feat: expose deterministic action_ref on ToolContext

src/agents/tool_context.py

@@ -1,11 +1,20 @@
 from __future__ import annotations
 
+import hashlib
+import json
 from dataclasses import dataclass, field, fields
 from typing import TYPE_CHECKING, Any, cast
 
 from openai.types.responses import ResponseFunctionToolCall
 
-from ._tool_identity import HostedMCPApprovalKey, get_tool_call_namespace, tool_trace_name
+from ._tool_identity import (
+    FunctionToolLookupKey,
+    HostedMCPApprovalKey,
+    get_function_tool_lookup_key,
+    get_tool_call_namespace,
+    serialize_function_tool_lookup_key,
+    tool_trace_name,
+)
 from ._tool_invocation import tool_invocation_identity, tool_invocation_identity_and_scope
 from .agent_tool_state import (
     get_agent_tool_state_scope,
@@ -35,9 +44,49 @@ def _assert_must_pass_tool_arguments() -> str:
     raise ValueError("tool_arguments must be passed to ToolContext")
 
 
+_ACTION_REF_DOMAIN = b"openai-agents-python:tool-action-ref:v1\0"
 _MISSING = object()
 
 
+def _canonical_tool_request(tool_arguments: str) -> bytes:
+    """Return deterministic bytes for hashing a tool request without changing execution behavior."""
+    try:
+        parsed_arguments = json.loads(tool_arguments)
+        canonical_arguments = json.dumps(
+            parsed_arguments,
+            ensure_ascii=True,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+    except (json.JSONDecodeError, TypeError, ValueError):
+        return b"raw\0" + tool_arguments.encode("utf-8", errors="surrogatepass")
+    return b"json\0" + canonical_arguments.encode("ascii")
+
+
+def _compute_tool_action_ref(
+    agent_name: str | None,
+    tool_lookup_key: FunctionToolLookupKey | None,
+    tool_arguments: str,
+) -> str | None:
+    if agent_name is None or tool_lookup_key is None:
+        return None
+
+    request_digest = hashlib.sha256(_canonical_tool_request(tool_arguments)).hexdigest()
+    commitment = json.dumps(
+        {
+            "agent_id": agent_name,
+            "request_digest": request_digest,
+            "tool_identity": serialize_function_tool_lookup_key(tool_lookup_key),
+        },
+        ensure_ascii=True,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("ascii")
+    digest = hashlib.sha256(_ACTION_REF_DOMAIN + commitment).hexdigest()
+    return f"act_v1_{digest}"
+
+
 @dataclass(eq=False)
 class ToolContext(RunContextWrapper[TContext]):
     """The context of a tool call."""
@@ -63,6 +112,8 @@ class ToolContext(RunContextWrapper[TContext]):
     run_config: RunConfig | None = None
     """The active run config for this tool call, when available."""
 
+    _action_ref: str | None = field(default=None, init=False, repr=False)
+
     def __init__(
         self,
         context: TContext,
@@ -114,6 +165,11 @@ def __init__(
             self.run_config = _coerce_run_config(run_config)
         else:
             self.run_config = None
+        self._action_ref = _compute_tool_action_ref(
+            agent.name if agent is not None else None,
+            get_function_tool_lookup_key(self.tool_name, self.tool_namespace),
+            self.tool_arguments,
+        )
         # Internal adapter hook used to attach SDK-only custom data to the emitted output item.
         self._custom_data: dict[str, Any] | None = None
 
@@ -122,6 +178,11 @@ def qualified_tool_name(self) -> str:
         """Return the tool name qualified by namespace when available."""
         return tool_trace_name(self.tool_name, self.tool_namespace) or self.tool_name
 
+    @property
+    def action_ref(self) -> str | None:
+        """Return the deterministic commitment for this tool action, when agent metadata exists."""
+        return self._action_ref
+
     def _find_nested_approval_target(
         self,
         approval_item: ToolApprovalItem,

tests/test_tool_context_action_ref.py

@@ -0,0 +1,144 @@
+import json
+from typing import Any
+
+import pytest
+
+from agents import Agent, RunHooks, Runner, function_tool
+from agents.run_context import RunContextWrapper
+from agents.testing import ScriptedModel
+from agents.tool import Tool
+from agents.tool_context import ToolContext
+
+from .test_responses import get_function_tool_call, get_text_message
+
+
+def _tool_context(
+    *,
+    agent_name: str = "Verifier",
+    tool_name: str = "lookup",
+    tool_arguments: str = '{"account": 7, "active": true}',
+    tool_namespace: str | None = None,
+) -> ToolContext[dict[str, object]]:
+    return ToolContext(
+        context={},
+        tool_name=tool_name,
+        tool_call_id="call-local",
+        tool_arguments=tool_arguments,
+        tool_namespace=tool_namespace,
+        agent=Agent(name=agent_name),
+    )
+
+
+def test_action_ref_is_stable_for_equivalent_json_arguments() -> None:
+    first = _tool_context(tool_arguments='{"account":7,"active":true}')
+    second = _tool_context(tool_arguments='{ "active": true, "account": 7 }')
+
+    assert first.action_ref is not None
+    assert first.action_ref.startswith("act_v1_")
+    assert first.action_ref == second.action_ref
+
+
+def test_action_ref_handles_json_surrogate_escape() -> None:
+    first = _tool_context(tool_arguments='{"x":"\\ud800"}')
+    second = _tool_context(tool_arguments='{ "x": "\\ud800" }')
+
+    assert first.action_ref is not None
+    assert first.action_ref == second.action_ref
+
+
+def test_action_ref_commits_agent_tool_and_request() -> None:
+    baseline = _tool_context().action_ref
+
+    assert baseline is not None
+    assert _tool_context(agent_name="Other verifier").action_ref != baseline
+    assert _tool_context(tool_name="update").action_ref != baseline
+    assert _tool_context(tool_arguments='{"account": 8, "active": true}').action_ref != baseline
+    assert _tool_context(tool_namespace="billing").action_ref != baseline
+
+
+def test_action_ref_distinguishes_bare_and_deferred_tool_identity() -> None:
+    bare = _tool_context(tool_name="lookup")
+    deferred = _tool_context(tool_name="lookup", tool_namespace="lookup")
+
+    assert bare.qualified_tool_name == deferred.qualified_tool_name == "lookup"
+    assert bare.action_ref is not None
+    assert deferred.action_ref is not None
+    assert bare.action_ref != deferred.action_ref
+
+
+def test_action_ref_does_not_depend_on_opaque_tool_call_id() -> None:
+    agent = Agent(name="Verifier")
+    first: ToolContext[dict[str, object]] = ToolContext(
+        context={},
+        tool_name="lookup",
+        tool_call_id="call-1",
+        tool_arguments='{"account": 7}',
+        agent=agent,
+    )
+    second: ToolContext[dict[str, object]] = ToolContext(
+        context={},
+        tool_name="lookup",
+        tool_call_id="call-2",
+        tool_arguments='{"account": 7}',
+        agent=agent,
+    )
+
+    assert first.action_ref == second.action_ref
+
+
+def test_action_ref_is_none_without_agent_metadata() -> None:
+    context: ToolContext[dict[str, object]] = ToolContext(
+        context={},
+        tool_name="lookup",
+        tool_call_id="call-1",
+        tool_arguments="{}",
+    )
+
+    assert context.action_ref is None
+
+
+class CaptureActionRefHooks(RunHooks[Any]):
+    def __init__(self) -> None:
+        self.action_refs: list[str | None] = []
+
+    async def on_tool_start(
+        self,
+        context: RunContextWrapper[Any],
+        agent: Agent[Any],
+        tool: Tool,
+    ) -> None:
+        assert isinstance(context, ToolContext)
+        self.action_refs.append(context.action_ref)
+
+    async def on_tool_end(
+        self,
+        context: RunContextWrapper[Any],
+        agent: Agent[Any],
+        tool: Tool,
+        result: object,
+    ) -> None:
+        assert isinstance(context, ToolContext)
+        self.action_refs.append(context.action_ref)
+
+
+@pytest.mark.asyncio
+async def test_action_ref_is_unchanged_across_tool_hooks() -> None:
+    @function_tool
+    def echo(value: str) -> str:
+        return value
+
+    hooks = CaptureActionRefHooks()
+    model = ScriptedModel()
+    model.extend(
+        [
+            [get_function_tool_call("echo", json.dumps({"value": "hello"}))],
+            [get_text_message("done")],
+        ]
+    )
+    agent = Agent(name="Hook verifier", model=model, tools=[echo])
+
+    await Runner.run(agent, input="call the echo tool", hooks=hooks)
+
+    assert len(hooks.action_refs) == 2
+    assert hooks.action_refs[0] is not None
+    assert hooks.action_refs[0] == hooks.action_refs[1]