Python: refresh agent-level run tools before each function-calling turn

docs/specs/004-python-function-calling-loop.md

@@ -591,7 +591,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | OpenAI hosted/local shell boundary | Hosted shell calls remain informational in streaming and non-streaming responses even when a local executor is configured; only valid explicit local-shell items or shell calls marked with a local environment can execute, local execution preserves its configured approval mode, stateless loops preserve the complete provider shell transcript or fail explicitly, and locally generated shell outputs are sent in every continuation mode. | `packages/openai/tests/openai/test_openai_chat_client.py::test_response_content_creation_with_shell_call_remains_hosted_with_local_tool`, `test_parse_chunk_from_openai_shell_call_done_remains_hosted`, `test_parse_chunk_from_openai_local_environment_shell_call_done_emits_command`, `test_mixed_shell_calls_only_invoke_explicit_local_shell_call`, `test_stateless_shell_transcript_without_provider_item_fails`, `test_prepare_messages_keeps_local_shell_output_under_storage`, `test_malformed_local_environment_shell_call_is_not_executable`, `test_response_content_creation_with_local_shell_call_maps_to_function_call`, `test_malformed_local_shell_call_is_not_executable`, `test_response_function_call_named_local_shell_is_informational`, `test_parse_chunk_function_call_named_local_shell_is_informational`, `test_local_shell_tool_requires_approval_before_function_loop_execution` |
 | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` |
 | Function invocation disabled | The client bypasses the invocation loop without losing invocation kwargs. | `test_function_invocation_config_enabled_false`, `test_function_invocation_config_enabled_false_preserves_invocation_kwargs`, `test_streaming_function_invocation_config_enabled_false` |
-| Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool` |
+| Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. In an agent run, tools appended to the agent's tools, its MCP servers, or a run-level tools list are sent on the next model call in streaming and non-streaming execution; tools removed through the invocation context stay removed, and a tool whose name is already taken is skipped with a warning on its own, without dropping other tools added in the same turn or aborting the run. Tools added mid-run are resolved in run-start source order (agent tools, run-level tools, then MCP servers), so the earlier source keeps a clashing name; at run start the same clash raises instead. Before each model call a source list is re-scanned only when it was replaced or its length changed, so an unchanged source costs no scan and an in-place replacement that keeps the same length takes effect from the next run. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool`, `packages/core/tests/core/test_agents.py::test_tool_appended_to_agent_tools_during_run_is_sent_on_next_request`, `test_tool_appended_to_run_tools_during_run_is_sent_on_next_request`, `test_mcp_server_added_to_agent_during_run_is_sent_on_next_request`, `test_tool_removed_with_context_during_run_is_not_restored_from_agent_tools`, `test_tool_with_duplicate_name_added_during_run_is_skipped`, `test_tools_list_identity_is_unchanged_across_iterations_when_nothing_is_added`, `test_same_tool_object_appended_twice_during_run_is_idempotent`, `test_valid_tool_added_with_duplicate_in_same_turn_is_still_exposed`, `test_two_new_tools_with_same_name_in_one_batch_keep_the_first`, `test_mcp_server_added_during_run_with_partially_clashing_functions_exposes_the_rest`, `test_local_tool_wins_over_mcp_function_appended_in_same_turn`, `test_tools_added_mid_run_match_run_start_tool_order`, `test_run_start_rejects_mcp_function_clashing_with_local_tool`, `test_refresh_does_not_scan_unchanged_tool_sources`, `test_disconnected_mcp_server_added_during_run_is_entered_discovered_and_closed` |
 | In-run compaction summaries | Summaries inserted for tool groups excluded by in-run compaction are returned in the final non-streaming response transcript before the group they replace, so history loaded with `skip_excluded` keeps the summarized content; summaries of caller-owned input messages are not added. | `packages/core/tests/core/test_clients.py::test_function_loop_returns_compaction_summaries_in_final_response`, `test_function_loop_returns_compaction_summaries_when_iteration_budget_exhausted`, `test_function_loop_reconciles_nested_compaction_summaries`, `test_function_loop_returns_compacted_transcript_on_early_terminal_exit`, `packages/core/tests/core/test_agents.py::test_agent_run_returns_and_persists_compaction_summaries` |
 | Per-service-call provider context | When framework-managed history is loaded around each model call, tools and instructions contributed by `HistoryProvider.before_run` remain available to the model and function loop without duplication in streaming or non-streaming execution. | `packages/core/tests/core/test_agents.py::test_vector_history_search_tool_is_available_with_per_service_call_persistence` |
 

python/packages/core/agent_framework/_agents.py

@@ -1481,6 +1481,44 @@ def _extract_conversation_id_from_streaming_response(
 
         return None
 
+    def _run_tool_sources(self, agent_tools: Any, run_tools: Any) -> list[Any]:
+        """Return a run's tool sources in discovery order: agent tools, run-level tools, then MCP servers.
+
+        Earlier sources win duplicate names at run start and among items discovered in the same refresh.
+        A name already exposed by an earlier refresh remains bound to its existing tool.
+        """
+        return [agent_tools, run_tools, self.mcp_tools]
+
+    def _run_tool_source_lists(self, run_tools: Any) -> list[list[Any] | None]:
+        """Return the tool lists a run re-reads before each model call, in precedence order.
+
+        These are the agent's tools, the run-level tools, and the agent's MCP servers. A source
+        that is not a list (for example a tuple or a single run-level tool) cannot grow during
+        the run and is returned as ``None``.
+        """
+        return [
+            cast("list[Any]", source) if isinstance(source, list) else None
+            for source in self._run_tool_sources(self.default_options.get("tools"), run_tools)
+        ]
+
+    async def _resolve_run_tool(
+        self,
+        tool: ToolTypes,
+        function_invocation_kwargs: Mapping[str, Any],
+    ) -> list[ToolTypes]:
+        """Resolve one normalized tool into the tools a run exposes, connecting MCP tools."""
+        from ._mcp import MCPTool
+
+        if not isinstance(tool, MCPTool):
+            return [tool]
+        await tool._prepare_for_run(function_invocation_kwargs)  # pyright: ignore[reportPrivateUsage]
+        if not tool.is_connected:
+            # The handshake and discovery requests are issued before any tool call, so the run's
+            # kwargs must reach header_provider here or those requests go out unauthenticated.
+            await self._async_exit_stack.enter_async_context(tool)
+            await tool._prepare_for_run(function_invocation_kwargs)  # pyright: ignore[reportPrivateUsage]
+        return list(tool.functions)
+
     async def _prepare_run_context(
         self,
         *,
@@ -1503,6 +1541,15 @@ async def _prepare_run_context(
         # remaining options into the request and silently override the resolved list.
         tools_ = _select_run_level_tools(tools, opts)
         opts.pop("tools", None)
+        # Items already in the agent's and the run's tool lists when the run starts; anything
+        # appended to those lists later is added to the run before its next model call. Each
+        # source's watermark (the list object and its length) lets an unchanged source be skipped
+        # without scanning its items.
+        start_source_lists = self._run_tool_source_lists(tools_)
+        seen_tool_items = {id(item): item for source in start_source_lists if source is not None for item in source}
+        source_watermarks: list[tuple[list[Any], int] | None] = [
+            (source, len(source)) if source is not None else None for source in start_source_lists
+        ]
 
         input_messages = normalize_messages(messages)
 
@@ -1586,10 +1633,13 @@ async def _prepare_run_context(
 
         agent_name = self._get_agent_name()
         from ._mcp import MCPTool
-        from ._tools import _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY  # pyright: ignore[reportPrivateUsage]
+        from ._tools import (
+            _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY,  # pyright: ignore[reportPrivateUsage]
+            _REFRESH_RUN_TOOLS_KEY,  # pyright: ignore[reportPrivateUsage]
+            FunctionInvocationLayer,
+        )
 
         base_tools = _normalize_tools(chat_options.pop("tools", None))
-        mcp_duplicate_message = "Tool names must be unique. Consider setting `tool_name_prefix` on the MCPTool."
 
         # Normalize tools
         normalized_tools = _normalize_tools(tools_)
@@ -1600,36 +1650,57 @@ async def _prepare_run_context(
         )
         additional_function_arguments = {**effective_function_invocation_kwargs, **existing_additional_args}
 
-        # Resolve final tool list (configured tools + runtime provided tools + local MCP server tools)
-        final_tools = list(base_tools)
-        for tool in normalized_tools:
-            if isinstance(tool, MCPTool):
-                await tool._prepare_for_run(additional_function_arguments)  # pyright: ignore[reportPrivateUsage]
-                if not tool.is_connected:
-                    # The handshake and discovery requests are issued before any tool call, so the run's
-                    # kwargs must reach header_provider here or those requests go out unauthenticated.
-                    await self._async_exit_stack.enter_async_context(tool)
-                    await tool._prepare_for_run(additional_function_arguments)  # pyright: ignore[reportPrivateUsage]
-                _append_unique_tools(
-                    final_tools,
-                    tool.functions,
-                    duplicate_error_message=mcp_duplicate_message,
-                )
-            else:
-                _append_unique_tools(final_tools, [tool])
-
-        for mcp_server in self.mcp_tools:
-            await mcp_server._prepare_for_run(additional_function_arguments)  # pyright: ignore[reportPrivateUsage]
-            if not mcp_server.is_connected:
-                await self._async_exit_stack.enter_async_context(mcp_server)
-                await mcp_server._prepare_for_run(  # pyright: ignore[reportPrivateUsage]
-                    additional_function_arguments
-                )
+        # Resolve final tool list in source precedence order. The agent's tools (including tools from
+        # context providers) are used as given; run-level tools and MCP servers are resolved after them.
+        agent_tools, *resolved_sources = self._run_tool_sources(base_tools, normalized_tools)
+        final_tools = list(agent_tools)
+        for tool in (tool for source in resolved_sources for tool in source):
             _append_unique_tools(
                 final_tools,
-                mcp_server.functions,
-                duplicate_error_message=mcp_duplicate_message,
+                await self._resolve_run_tool(tool, additional_function_arguments),
+                duplicate_error_message=(
+                    "Tool names must be unique. Consider setting `tool_name_prefix` on the MCPTool."
+                    if isinstance(tool, MCPTool)
+                    else None
+                ),
             )
+        # MCP tools added later in the run are prepared with the same kwargs as those added above.
+        mcp_run_kwargs = dict(additional_function_arguments)
+
+        async def refresh_run_tools() -> list[ToolTypes]:
+            # Candidates are returned without de-duplication; the function-invocation loop merges
+            # them one at a time and skips only the names that clash.
+            candidates: list[ToolTypes] = []
+            for index, source in enumerate(self._run_tool_source_lists(tools_)):
+                watermark = source_watermarks[index]
+                # Same list object and length as last time: nothing was appended, so skip the scan.
+                if source is None or (watermark is not None and watermark[0] is source and watermark[1] == len(source)):
+                    continue
+                source_resolved = True
+                for item in source:
+                    if id(item) in seen_tool_items:
+                        continue
+                    try:
+                        resolved = [
+                            resolved_tool
+                            for tool in _normalize_tools(item)
+                            for resolved_tool in await self._resolve_run_tool(tool, mcp_run_kwargs)
+                        ]
+                    except Exception:
+                        # Left unseen, and the watermark is not advanced, so resolution is retried
+                        # before the next model call.
+                        logger.warning(
+                            "Tool %r added during the run could not be resolved; retrying before the next model call.",
+                            item,
+                            exc_info=True,
+                        )
+                        source_resolved = False
+                        continue
+                    seen_tool_items[id(item)] = item
+                    candidates.extend(resolved)
+                if source_resolved:
+                    source_watermarks[index] = (source, len(source))
+            return candidates
 
         additional_function_arguments[_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY] = _tool_approval_source_ids(
             self.middleware
@@ -1678,6 +1749,9 @@ async def _prepare_run_context(
         if active_session is not None:
             effective_client_kwargs["session"] = active_session
             effective_client_kwargs[_APPROVAL_SESSION_IS_AUTHORITATIVE_KEY] = not framework_created_session
+        effective_client_kwargs.pop(_REFRESH_RUN_TOOLS_KEY, None)
+        if isinstance(self.client, FunctionInvocationLayer):
+            effective_client_kwargs[_REFRESH_RUN_TOOLS_KEY] = refresh_run_tools
         per_service_call_history_middleware: PerServiceCallHistoryPersistingMiddleware | None = None
         if per_service_call_history_providers and active_session is not None:
             per_service_call_history_middleware = PerServiceCallHistoryPersistingMiddleware(