Python: progressive tool exposure via FunctionInvocationContext

python/packages/core/AGENTS.md

@@ -56,7 +56,7 @@ agent_framework/
 - **`AgentMiddleware`** - Intercepts agent `run()` calls
 - **`ChatMiddleware`** - Intercepts chat client `get_response()` calls
 - **`FunctionMiddleware`** - Intercepts function/tool invocations
-- **`AgentContext`** / **`ChatContext`** / **`FunctionInvocationContext`** - Context objects passed through middleware
+- **`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).
 
 ### Sessions (`_sessions.py`)
 

python/packages/core/agent_framework/_feature_stage.py

@@ -57,6 +57,7 @@ class ExperimentalFeature(str, Enum):
     FOUNDRY_PREVIEW_TOOLS = "FOUNDRY_PREVIEW_TOOLS"
     FUNCTIONAL_WORKFLOWS = "FUNCTIONAL_WORKFLOWS"
     HARNESS = "HARNESS"
+    PROGRESSIVE_TOOLS = "PROGRESSIVE_TOOLS"
     SKILLS = "SKILLS"
     TO_PROMPT_AGENT = "TO_PROMPT_AGENT"
 

python/packages/core/agent_framework/_middleware.py

@@ -11,6 +11,7 @@
 from typing import TYPE_CHECKING, Any, Generic, Literal, TypeAlias, cast, overload
 
 from ._clients import SupportsChatGetResponse
+from ._feature_stage import ExperimentalFeature, experimental
 from ._types import (
     AgentResponse,
     AgentResponseUpdate,
@@ -214,6 +215,12 @@ class FunctionInvocationContext:
         result: Function execution result. Can be observed after calling ``call_next()``
                 to see the actual execution result or can be set to override the execution result.
         kwargs: Additional runtime keyword arguments forwarded to the function invocation.
+        tools: The live, mutable list of tools available to the model for the current
+                agent run, or ``None`` when the function is invoked outside of a
+                function-calling loop (for example via ``FunctionTool.invoke`` directly).
+                Tools can add or remove tools during execution using :meth:`add_tools`
+                and :meth:`remove_tools` (progressive tool exposure). Mutations take
+                effect on the **next** model iteration, not the in-flight batch.
 
     Examples:
         .. code-block:: python
@@ -232,6 +239,18 @@ async def process(self, context: FunctionInvocationContext, call_next):
 
                     # Continue execution
                     await call_next()
+
+        Progressive tool exposure from inside a tool:
+
+        .. code-block:: python
+
+            from agent_framework import FunctionInvocationContext, tool
+
+
+            @tool(approval_mode="never_require")
+            def load_math_tools(ctx: FunctionInvocationContext) -> str:
+                ctx.add_tools([factorial, fibonacci])
+                return "Math tools are now available."
     """
 
     def __init__(
@@ -242,6 +261,7 @@ def __init__(
         metadata: Mapping[str, Any] | None = None,
         result: Any = None,
         kwargs: Mapping[str, Any] | None = None,
+        tools: list[ToolTypes] | None = None,
     ) -> None:
         """Initialize the FunctionInvocationContext.
 
@@ -252,13 +272,105 @@ def __init__(
             metadata: Metadata dictionary for sharing data between function middleware.
             result: Function execution result.
             kwargs: Additional runtime keyword arguments forwarded to the function invocation.
+            tools: The live, mutable list of tools for the current agent run. When provided,
+                this is the same list object the model sees on the next iteration, so
+                appending or removing tools changes the model's available tools.
         """
         self.function = function
         self.arguments = arguments
         self.session = session
         self.metadata: dict[str, Any] = dict(metadata) if metadata is not None else {}
         self.result = result
         self.kwargs: dict[str, Any] = dict(kwargs) if kwargs is not None else {}
+        self.tools = tools
+
+    @experimental(feature_id=ExperimentalFeature.PROGRESSIVE_TOOLS)
+    def add_tools(
+        self,
+        tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]],
+    ) -> None:
+        """Add one or more tools to the current agent run (progressive tool exposure).
+
+        Callable inputs are converted to :class:`FunctionTool`, and tool collections are
+        flattened, using the same normalization as the rest of the framework. Added tools
+        become available to the model on the **next** iteration of the function-calling
+        loop; they do not affect tool calls already requested in the in-flight batch.
+
+        Adding a tool whose name already exists is a no-op when it is the same object, and
+        raises ``ValueError`` when it is a different object with a duplicate name.
+
+        Args:
+            tools: A single tool/callable or a sequence of tools/callables to add.
+
+        Raises:
+            RuntimeError: If the context has no live tools list (for example when the
+                function is invoked outside of a function-calling loop).
+            ValueError: If a different tool with a duplicate name is added.
+        """
+        from ._tools import _append_unique_tools, normalize_tools  # type: ignore[reportPrivateUsage]
+
+        if self.tools is None:
+            raise RuntimeError(
+                "Cannot add tools: this FunctionInvocationContext is not bound to a live "
+                "agent run. add_tools is only available for functions invoked within an "
+                "agent's function-calling loop."
+            )
+        # Validate the whole batch against a throwaway copy first, so a duplicate-name
+        # clash partway through the batch raises before the live tool list is mutated
+        # (all-or-nothing semantics).
+        merged = _append_unique_tools(list(self.tools), normalize_tools(tools))
+        self.tools[:] = merged
+
+    def remove_tools(
+        self,
+        tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | str | Sequence[str],
+    ) -> None:
+        """Remove one or more tools from the current agent run (progressive tool exposure).
+
+        Tools may be specified by name, by tool object, or by the original callable. Names
+        that are not currently present are ignored. Removals take effect on the **next**
+        iteration of the function-calling loop; tool calls already requested in the
+        in-flight batch still execute.
+
+        Args:
+            tools: A tool name, tool/callable, or a sequence of any of these to remove.
+
+        Raises:
+            RuntimeError: If the context has no live tools list (for example when the
+                function is invoked outside of a function-calling loop).
+        """
+        from ._tools import _get_tool_name, normalize_tools  # type: ignore[reportPrivateUsage]
+
+        if self.tools is None:
+            raise RuntimeError(
+                "Cannot remove tools: this FunctionInvocationContext is not bound to a live "
+                "agent run. remove_tools is only available for functions invoked within an "
+                "agent's function-calling loop."
+            )
+
+        names_to_remove: set[str] = set()
+        raw_items: list[Any]
+        if isinstance(tools, str):
+            raw_items = [tools]
+        elif isinstance(tools, Sequence) and not isinstance(tools, (bytes, bytearray)):
+            raw_items = list(cast("Sequence[Any]", tools))
+        else:
+            raw_items = [tools]
+        for item in raw_items:
+            if isinstance(item, str):
+                names_to_remove.add(item)
+                continue
+            for normalized in normalize_tools(item):
+                if name := _get_tool_name(normalized):  # type: ignore[reportPrivateUsage]
+                    names_to_remove.add(name)
+
+        if not names_to_remove:
+            return
+        self.tools[:] = [
+            tool
+            for tool in self.tools
+            if _get_tool_name(tool) not in names_to_remove  # type: ignore[reportPrivateUsage]
+        ]
 
 
 class ChatContext:

python/packages/core/agent_framework/_tools.py

@@ -1418,6 +1418,7 @@ async def _auto_invoke_function(
     sequence_index: int | None = None,
     request_index: int | None = None,
     middleware_pipeline: FunctionMiddlewarePipeline | None = None,
+    live_tools: list[ToolTypes] | None = None,
 ) -> Content:
     """Invoke a function call requested by the agent, applying middleware that is defined.
 
@@ -1432,6 +1433,8 @@ async def _auto_invoke_function(
         sequence_index: The index of the function call in the sequence.
         request_index: The index of the request iteration.
         middleware_pipeline: Optional middleware pipeline to apply during execution.
+        live_tools: The live, mutable tools list for the current agent run, exposed on
+            the FunctionInvocationContext so tools can add/remove tools at runtime.
 
     Returns:
         The function result content.
@@ -1523,6 +1526,7 @@ async def _auto_invoke_function(
                     arguments=args,
                     session=invocation_session,
                     kwargs=runtime_kwargs.copy(),
+                    tools=live_tools,
                 )
             function_result = await tool.invoke(
                 arguments=args,
@@ -1537,6 +1541,12 @@ async def _auto_invoke_function(
         except UserInputRequiredException:
             raise
         except Exception as exc:
+            logger.warning(
+                "Function '%s' raised an exception; returning an error result to the model. "
+                "Set include_detailed_errors=True or inspect this log for details.",
+                tool.name,
+                exc_info=exc,
+            )
             message = "Error: Function failed."
             if config.get("include_detailed_errors", False):
                 message = f"{message} Exception: {exc}"
@@ -1552,6 +1562,7 @@ async def _auto_invoke_function(
         arguments=args,
         session=invocation_session,
         kwargs=runtime_kwargs.copy(),
+        tools=live_tools,
     )
 
     call_id = function_call_content.call_id
@@ -1608,6 +1619,12 @@ async def final_function_handler(context_obj: Any) -> Any:
     except UserInputRequiredException:
         raise
     except Exception as exc:
+        logger.warning(
+            "Function '%s' raised an exception; returning an error result to the model. "
+            "Set include_detailed_errors=True or inspect this log for details.",
+            tool.name,
+            exc_info=exc,
+        )
         message = "Error: Function failed."
         if config.get("include_detailed_errors", False):
             message = f"{message} Exception: {exc}"
@@ -1659,6 +1676,9 @@ async def _try_execute_function_calls(
     from ._types import Content
 
     tool_map = _get_tool_map(tools)
+    # The live tools list (when tools is the run-local list) is exposed on the
+    # FunctionInvocationContext so tools can add/remove tools during the run.
+    live_tools: list[ToolTypes] | None = cast("list[ToolTypes]", tools) if isinstance(tools, list) else None
     approval_tools = [tool_name for tool_name, tool in tool_map.items() if tool.approval_mode == "always_require"]
     logger.debug(
         "_try_execute_function_calls: tool_map keys=%s, approval_tools=%s",
@@ -1733,6 +1753,7 @@ async def invoke_with_termination_handling(
                 request_index=attempt_idx,
                 middleware_pipeline=middleware_pipeline,
                 config=config,
+                live_tools=live_tools,
             )
             return (result, False)
         except MiddlewareTermination as exc:
@@ -2371,6 +2392,13 @@ def get_response(
                 function_invocation_kwargs=function_invocation_kwargs,
                 client_kwargs=filtered_kwargs,
             )
+        # Establish a single, run-local mutable tools list so that tools can add or remove
+        # tools during the run (progressive tool exposure). A fresh list is created via
+        # normalize_tools so the caller's original tools container is never mutated, while
+        # the same list object is shared with the model (options["tools"]) and the tool map
+        # rebuilt on every loop iteration.
+        if mutable_options.get("tools"):
+            mutable_options["tools"] = normalize_tools(mutable_options["tools"])
         if not stream:
 
             async def _get_response() -> ChatResponse[Any]: