Daniel Quintas (danielquintas8) · Nov 3, 2026 · Nov 4, 2026
@@ -425,6 +425,11 @@ that manually replay messages own the equivalent rule: do not resend an approval
 - Parallel calls retain model order in the returned transcript.
 - `call_id` remains the provider/service correlation id; a locally actionable `function_call` also carries a stable
   Agent Framework occurrence identity in `Content.id`.
+- `FunctionInvocationContext.call_id` exposes that provider ID as a read-only property before function middleware
+  and tool execution, with or without middleware and after approval resume. Standalone `FunctionTool.invoke` calls
+  use the supplied `tool_call_id`, or `None` when omitted; reusing a context does not retain an earlier call ID.
+  Existing middleware identity metadata remains available. This property does not expose the call's payload or
+  framework occurrence identity.
 - Reused `call_id` values are correlated by logical occurrence, not one global value per id.
 - Existing `Content.id` values survive aggregation and replay and are never regenerated; legacy deserialization does
   not invent one.
@@ -605,6 +610,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | String input | Flexible string input follows the same loop behavior. | `test_base_client_with_function_calling_string_input` |
 | Multiple sequential rounds | Each round retains one call/result pair. | `test_base_client_with_function_calling_resets` |
 | Streaming call | Call chunks, one result update, and final text are emitted in order. | `test_base_client_with_streaming_function_calling` |
+| Tool-visible provider call ID | Tools and function middleware can read the provider ID on both execution paths, in both response modes, and after approval resume. The public property is read-only. Standalone invocation uses its supplied ID or `None`, including when a context is reused. | `test_tool_context_exposes_provider_call_id`, `packages/core/tests/core/test_tools.py::test_invoke_exposes_tool_call_id_to_context`, `test_invoke_clears_call_id_when_reusing_context_without_an_id`, `packages/core/tests/core/test_middleware.py::TestFunctionInvocationContext` |
 | Function-call occurrence identity | Actionable calls gain one stable `Content.id`; a safe local empty-`call_id` fallback uses that id with a migration warning, and streaming aggregation preserves provider-assigned occurrence ids across interleaved fragments. OpenAI Chat Completions scopes fragment correlation to each request and `(choice.index, tool.index)`. | `test_actionable_function_call_gets_stable_occurrence_identity`, `test_actionable_function_call_uses_occurrence_identity_for_empty_call_id`, `test_streaming_empty_call_id_keeps_occurrence_identity_through_approval`, `test_streaming_empty_call_id_delta_reuses_opening_call_identity`, `test_streaming_interleaved_indexed_call_fragments_coalesce_by_occurrence`, `packages/core/tests/core/test_types.py::test_function_call_occurrence_id_roundtrips_without_regeneration`, `packages/openai/tests/openai/test_openai_chat_completion_client.py::test_streaming_tool_call_identity_is_request_local_and_scoped_by_choice_index` |
 | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` |
 | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` |

@@ -195,6 +195,9 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID
 - **`AgentMiddleware`** - Intercepts agent `run()` calls
 - **`ChatMiddleware`** - Intercepts chat client `get_response()` calls
 - **`FunctionMiddleware`** - Intercepts function/tool invocations
+- **`FunctionInvocationContext.call_id`** - Read-only provider call ID, available before function middleware and
+  tool execution in the automatic loop, including approval resume. Standalone `FunctionTool.invoke` calls use
+  `tool_call_id` and expose `None` when it is omitted. Existing middleware identity metadata remains available.
 - **`AgentContext`** / **`ChatContext`** / **`FunctionInvocationContext`** - Context objects passed through middleware. A tool can declare a `FunctionInvocationContext` parameter to receive it; `context.tools` is the live, mutable tools list for the run, and `context.add_tools(...)` / `context.remove_tools(...)` enable progressive tool exposure (changes apply on the next function-calling iteration). Chat middleware that constructs provider-local replacement messages must call `context.record_message_replacement(...)` for every replacement before downstream compaction so complete summaries can reconcile to caller-owned messages without persisting the replacements themselves.
 - **`MessageInjectionMiddleware`** - Session-scoped chat middleware that lets tools or other code enqueue messages for the next model call in the current `AgentSession`; it drains queued messages into the next call and loops only when no function calls need to be handled by the function invocation layer.
 

@@ -439,6 +439,11 @@ class FunctionInvocationContext:
 
     Attributes:
         function: The function being invoked.
+        call_id: The read-only provider call ID for this invocation. Automatic invocation
+                supplies it before function middleware and tool execution, including
+                approval resume. Direct ``FunctionTool.invoke`` calls use ``tool_call_id``;
+                it is ``None`` when no ID is supplied. This is distinct from the
+                framework occurrence identity in ``Content.id``.
         arguments: The function arguments. In the automatic function-calling loop,
                 schema-compatible provider arguments retain the existing normalized
                 mapping contract. If provisional normalization rejects provider
@@ -535,6 +540,7 @@ def __init__(
                 or ``None`` when the invocation path cannot establish that ownership.
         """
         self.function = function
+        self._call_id: str | None = None
         self.arguments = arguments
         self.session = session
         self.metadata: dict[str, Any] = dict(metadata) if metadata is not None else {}
@@ -545,6 +551,11 @@ def __init__(
         )
         self.tools = tools
 
+    @property
+    def call_id(self) -> str | None:
+        """Get the provider call ID supplied by the invocation layer, if any."""
+        return self._call_id
+
     @experimental(feature_id=ExperimentalFeature.PROGRESSIVE_TOOLS)
     def add_tools(
         self,

@@ -1153,6 +1153,9 @@ async def invoke(
         Raises:
             TypeError: If arguments is not mapping-like or fails schema checks.
         """
+        if context is not None:
+            # Bind before validation so failed invocations cannot retain an earlier call ID.
+            context._call_id = tool_call_id  # pyright: ignore[reportPrivateUsage]
         if self.declaration_only:
             raise ToolException(f"Function '{self.name}' is declaration only and cannot be invoked.")
         global OBSERVABILITY_SETTINGS
@@ -1196,6 +1199,7 @@ async def invoke(
                 arguments=validated_arguments,
                 kwargs=runtime_kwargs,
             )
+            effective_context._call_id = tool_call_id  # pyright: ignore[reportPrivateUsage]
         if effective_context is not None:
             effective_context.function = self
             effective_context.arguments = validated_arguments
@@ -2395,6 +2399,7 @@ async def _auto_invoke_function(
         raise KeyError(f'Function "{function_call_content.name}" is missing call_id.')
 
     # Pass both provider correlation and framework occurrence identity to middleware.
+    middleware_context._call_id = call_id  # pyright: ignore[reportPrivateUsage]  # Bind before middleware runs.
     middleware_context.metadata["call_id"] = call_id
     if function_call_content.id is not None:
         middleware_context.metadata["function_call_occurrence_id"] = function_call_content.id

@@ -971,6 +971,130 @@ def _finalize(stream_updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
     chat_client_base._get_streaming_response = _get_streaming_response
 
 
+@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
+@pytest.mark.parametrize("with_middleware", [False, True], ids=["direct", "middleware"])
+@pytest.mark.parametrize("requires_approval", [False, True], ids=["immediate", "approved"])
+async def test_tool_context_exposes_provider_call_id(
+    chat_client_base: SupportsChatGetResponse,
+    streaming: bool,
+    with_middleware: bool,
+    requires_approval: bool,
+) -> None:
+    """Tools and middleware receive the provider ID across execution and approval paths."""
+    observed_call_ids: list[str] = []
+    middleware_call_ids: list[str] = []
+
+    class ObserveCall(FunctionMiddleware):
+        async def process(self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]) -> None:
+            assert context.call_id is not None
+            assert context.metadata["call_id"] == context.call_id
+            middleware_call_ids.append(context.call_id)
+            await call_next()
+
+    @tool(name="lookup", approval_mode="always_require" if requires_approval else "never_require")
+    async def lookup(context: FunctionInvocationContext, value: str) -> str:
+        await asyncio.sleep(0)
+        provider_call_id = context.call_id
+        assert provider_call_id is not None
+        observed_call_ids.append(provider_call_id)
+        return value
+
+    calls = [
+        Content.from_function_call(
+            call_id=f"provider-{index}",
+            name="lookup",
+            arguments='{"value":"same"}',
+            id=f"occurrence-{index}",
+            additional_properties={"provider": {"item_id": f"item-{index}"}},
+        )
+        for index in (1, 2)
+    ]
+    responses = [
+        ChatResponse(messages=Message(role="assistant", contents=calls)),
+        ChatResponse(messages=Message(role="assistant", contents=["done"])),
+    ]
+    if streaming:
+        chat_client_base.streaming_responses = [  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+            [
+                ChatResponseUpdate(
+                    role="assistant",
+                    contents=[
+                        Content.from_function_call(
+                            call_id=f"provider-{index}",
+                            name="lookup",
+                            arguments='{"value":',
+                            id=f"occurrence-{index}",
+                            additional_properties={"provider": {"item_id": f"item-{index}"}},
+                        )
+                        for index in (1, 2)
+                    ],
+                ),
+                ChatResponseUpdate(
+                    role="assistant",
+                    contents=[
+                        Content.from_function_call(
+                            call_id=f"provider-{index}", name="", arguments='"same"}', id=f"occurrence-{index}"
+                        )
+                        for index in (1, 2)
+                    ],
+                ),
+            ],
+            [ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])],
+        ]
+    else:
+        chat_client_base.run_responses = responses  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+
+    agent = Agent(client=chat_client_base, tools=[lookup], middleware=[ObserveCall()] if with_middleware else [])
+    session = agent.create_session()
+
+    async def run(messages: str | list[Message]) -> list[Message]:
+        if streaming:
+            response_stream = agent.run(messages, session=session, stream=True)
+            updates = [update async for update in response_stream]
+            response = await response_stream.get_final_response()
+            assert [
+                content.call_id
+                for update in updates
+                for content in update.contents
+                if content.type == "function_result"
+            ] == [
+                content.call_id
+                for message in response.messages
+                for content in message.contents
+                if content.type == "function_result"
+            ]
+        else:
+            response = await agent.run(messages, session=session)
+        return response.messages
+
+    messages = await run("look up the same value twice")
+    if requires_approval:
+        assert observed_call_ids == []
+        assert middleware_call_ids == []
+        requests = [
+            content
+            for message in messages
+            for content in message.contents
+            if content.type == "function_approval_request"
+        ]
+        assert len(requests) == 2
+        session = AgentSession.from_dict(json.loads(json.dumps(session.to_dict())))
+        messages = await run([
+            Message(
+                role="user", contents=[request.to_function_approval_response(approved=True) for request in requests]
+            )
+        ])
+
+    assert sorted(observed_call_ids) == ["provider-1", "provider-2"]
+    assert middleware_call_ids == (["provider-1", "provider-2"] if with_middleware else [])
+    assert [
+        (content.call_id, content.id, content.result)
+        for message in messages
+        for content in message.contents
+        if content.type == "function_result"
+    ] == [("provider-1", "occurrence-1", "same"), ("provider-2", "occurrence-2", "same")]
+
+
 async def test_base_client_with_function_calling(chat_client_base: SupportsChatGetResponse):
     exec_counter = 0
 

@@ -99,6 +99,16 @@ def test_init_with_defaults(self, mock_function: FunctionTool) -> None:
         assert context.function is mock_function
         assert context.arguments == arguments
         assert context.metadata == {}
+        assert context.call_id is None
+
+    def test_call_id_is_read_only(self, mock_function: FunctionTool) -> None:
+        """Applications cannot overwrite the framework-owned call ID through the public property."""
+        context = FunctionInvocationContext(function=mock_function, arguments={})
+
+        with pytest.raises(AttributeError):
+            setattr(context, "call_id", "replacement")  # noqa: B010 - Exercise runtime rejection, not typed assignment.
+
+        assert context.call_id is None
 
     def test_init_with_custom_metadata(self, mock_function: FunctionTool) -> None:
         """Test FunctionInvocationContext initialization with custom metadata."""

@@ -21,7 +21,7 @@
     FunctionTool,
     tool,
 )
-from agent_framework._middleware import FunctionInvocationContext
+from agent_framework._middleware import FunctionInvocationContext, MiddlewareFailure
 from agent_framework._tools import (
     _auto_invoke_function,
     _format_tool_parameters,
@@ -1608,6 +1608,98 @@ def tool_with_kwargs(x: int, **kwargs: Any) -> str:
     assert result_default[0].text == "x=10, user=unknown"
 
 
+@pytest.mark.parametrize("explicit_context", [False, True])
+@pytest.mark.parametrize("tool_call_id", [None, "manual-call"])
+async def test_invoke_exposes_tool_call_id_to_context(explicit_context: bool, tool_call_id: str | None) -> None:
+    """Standalone invocation exposes the supplied ID without fabricating one."""
+    observed_call_ids: list[str | None] = []
+
+    @tool(approval_mode="never_require")
+    def lookup(context: FunctionInvocationContext) -> str:
+        observed_call_ids.append(context.call_id)
+        return "found"
+
+    context = FunctionInvocationContext(function=lookup, arguments={}) if explicit_context else None
+    result = await lookup.invoke(context=context, tool_call_id=tool_call_id)
+
+    assert observed_call_ids == [tool_call_id]
+    assert result[0].text == "found"
+    assert lookup.parameters()["properties"] == {}
+
+
+async def test_invoke_clears_call_id_when_reusing_context_without_an_id() -> None:
+    """A reused manual context must not retain the previous invocation's call ID."""
+    observed_call_ids: list[str | None] = []
+
+    @tool(approval_mode="never_require")
+    def lookup(context: FunctionInvocationContext) -> str:
+        observed_call_ids.append(context.call_id)
+        return "found"
+
+    context = FunctionInvocationContext(function=lookup, arguments={})
+    await lookup.invoke(context=context, tool_call_id="first-call")
+    await lookup.invoke(context=context)
+
+    assert observed_call_ids == ["first-call", None]
+
+
+@pytest.mark.parametrize("tool_call_id", [None, "second-call"])
+@pytest.mark.parametrize(
+    ("arguments", "invocation_kwargs", "error_message"),
+    [
+        ({"item": "invalid"}, {}, "Invalid arguments"),
+        ({"item": 1}, {"unexpected": True}, "Unexpected keyword argument"),
+    ],
+)
+async def test_invoke_updates_call_id_before_argument_validation(
+    tool_call_id: str | None,
+    arguments: dict[str, Any],
+    invocation_kwargs: dict[str, Any],
+    error_message: str,
+) -> None:
+    """Rejected arguments must not leave an earlier call ID on a reused context."""
+    executed_items: list[int] = []
+
+    @tool(approval_mode="never_require")
+    def lookup(item: int) -> str:
+        executed_items.append(item)
+        return "found"
+
+    context = FunctionInvocationContext(function=lookup, arguments={"item": 1})
+    await lookup.invoke(context=context, tool_call_id="first-call")
+    assert context.call_id == "first-call"
+
+    with pytest.raises(TypeError, match=error_message):
+        await lookup.invoke(arguments=arguments, context=context, tool_call_id=tool_call_id, **invocation_kwargs)
+
+    assert context.call_id == tool_call_id
+    assert executed_items == [1]
+
+
+@pytest.mark.parametrize("tool_call_id", [None, "second-call"])
+async def test_invoke_updates_call_id_before_security_validation(tool_call_id: str | None) -> None:
+    """Security rejection must preserve the failed invocation's ID without executing it."""
+    executed_items: list[int] = []
+
+    @tool(approval_mode="never_require")
+    def lookup(item: int) -> str:
+        executed_items.append(item)
+        return "found"
+
+    context = FunctionInvocationContext(function=lookup, arguments={"item": 1})
+    context.metadata[tools_module._SECURITY_ARGUMENTS_SNAPSHOT_CONTEXT_KEY] = tools_module._argument_authority_token(
+        {"item": 1}, boundary="security policy"
+    )
+    await lookup.invoke(context=context, tool_call_id="first-call")
+    assert context.call_id == "first-call"
+
+    with pytest.raises(MiddlewareFailure, match="Function arguments changed after security middleware"):
+        await lookup.invoke(arguments={"item": 2}, context=context, tool_call_id=tool_call_id)
+
+    assert context.call_id == tool_call_id
+    assert executed_items == [1]
+
+
 async def test_ai_function_with_explicit_invocation_context():
     """Test that invoke() can receive runtime kwargs via FunctionInvocationContext."""