diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index 0052716eaa1..f06cec0eda6 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -253,6 +253,15 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID `ClientSession` combinations remain compatible until their vNext removal. - **Argument allowlist (`_prepare_call_kwargs`)** - Before each `tools/call`, kwargs are filtered to an **allowlist** built from the tool's declared parameters (`inputSchema.properties`) plus any user-configured extras. **The declared half comes from the server's advertised schema**, and runtime kwargs (`FunctionInvocationContext.kwargs`, seeded from `function_invocation_kwargs`) are merged with the model-supplied arguments upstream in `_call_tool_with_runtime_kwargs`, so provenance is gone by the time the filter runs. A runtime kwarg is therefore forwarded whenever the server declares a property of that name, without the model mentioning it — the server, not the caller, decides which runtime kwarg names it receives. A tool that declares no usable `properties` (including schemas with `additionalProperties: true`) forwards only the configured extras. `_MCP_FRAMEWORK_DENYLIST` is a narrow safety net covering only non-serializable framework objects a server *declares* in its schema (those are dropped); it does not generalize to arbitrary caller-chosen names, and explicit extras always win. The reserved `_meta` key is never forwarded as an argument; trusted caller/runtime `_meta` is validated as MCP request metadata, model-supplied `_meta` is discarded in generated MCP functions, and metadata precedence is caller/runtime < OpenTelemetry < tools/list metadata. - **`allowed_tools`** (constructor arg on all `MCPTool` subclasses) - Restricts exposed MCP tools by raw remote MCP tool identity. Prefixed local names remain accepted only when the raw remote name already matches its normalized form; normalized/local aliases do not authorize a different raw remote name. Configured allow/approval names must identify at most one raw remote name across loaded tools and prompts: a raw name that overlaps another tool's prefixed alias raises `ToolExecutionException`. Discovery validates all pages before publishing new functions; a failed reload retains the previous functions and metadata. The allowlist is also revalidated when exposing functions so runtime changes cannot select an ambiguous name. If multiple raw remote tool names map to the same local function name, tool loading raises `ToolExecutionException` instead of first-one-wins shadowing. +- **`approval_mode`** (constructor arg on all `MCPTool` subclasses) - Accepts `"always_require"`, `"never_require"`, + `None` (no approval, the `FunctionTool` default), or an `MCPSpecificApproval` dict. A dict resolves each tool or prompt + in order: `always_require_approval`, then `never_require_approval`, then the optional `default` + (`"always_require"` or `"never_require"`, validated at construction; `None` means not set). Without `default`, a tool + or prompt named in neither list does not require approval, and setting `never_require_approval` without `default` + emits a `FutureWarning` because that fallback may change in a future release. A name list given as a bare string is one + name, not a substring pattern. `default` is local-only: hosted MCP helpers (`get_mcp_tool`) do not read it. Tools and + prompts discovered later (for example after `tools/list_changed`) use the policy at discovery time; retained + functions keep their resolved mode. - **Progressive MCP disclosure** (`use_progressive_disclosure`, `always_load`) - When enabled on any `MCPTool` subclass, the initial model-facing surface is loader tools (`list_mcp_tools` / `load_tool` / `unload_tool`, prefixed by `tool_name_prefix` when configured) plus allowed tools selected by `always_load` and tools loaded earlier on the same `MCPTool` instance. `list_mcp_tools` only reports tools that pass `allowed_tools`; filtered tools are not listed or loadable. Loader tool names are reserved in progressive mode: remote MCP tools whose local generated name collides with a loader name are omitted from the initial/listed surface, and explicit `load_tool` calls return a model-visible message pointing callers to `tool_name_prefix` or excluding the colliding tool. `load_tool` accepts one tool name or a list of tool names and uses `FunctionInvocationContext.add_tools(...)` so the selected generated MCP `FunctionTool`s become available on the next function-calling iteration while keeping existing approval mode, argument filtering, header-provider runtime kwargs, result parsing, and OTel. `unload_tool` accepts one dynamically loaded tool name or a list of names and removes them from the live tool list and persisted progressive surface, but it does not remove tools configured in `always_load`. Invalid `always_load` entries are ignored like unmatched `allowed_tools` entries. - **Tool discovery refresh** - Validate the complete current tool snapshot alongside retained non-tool functions, not stale tool wrappers. Successful refreshes atomically replace the previous tool set and metadata, including empty snapshots, while preserving prompts, custom functions, and unchanged tool wrappers (including caller customizations). Removed or replaced tool identities lose their persisted progressive-loaded state; failed refreshes preserve the previous state. - **`additional_tool_argument_names`** (constructor arg on all `MCPTool` subclasses) - Opt extra argument names back into the allowlist. Accepts a `Sequence[str]` (applied to every tool) or a `Mapping[str, Sequence[str]]` keyed by **remote tool name**, where the reserved key `"*"` denotes global extras. It is configured only in user code at construction; there is **no per-call/runtime override**, so a model-issued tool call cannot change which names pass through — but note this constrains the *model*, not the *server*, which still widens the effective allowlist through its schema. To use a server that accepts `additionalProperties: true`, list the extra names here and then either (1) manually extend that tool's `inputSchema` (via the `.functions` list after connecting) so the model is prompted to supply them, or (2) supply the values yourself via `function_invocation_kwargs`. If a normal forwarded argument name is supplied by both the model and `function_invocation_kwargs`, the model-supplied value wins; `_meta` is the exception and only trusted runtime/caller metadata is used. diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index 37b4ef56cba..45b514cccf9 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -17,7 +17,7 @@ from dataclasses import dataclass from datetime import date, datetime from http.cookiejar import CookieJar, DefaultCookiePolicy -from inspect import isawaitable +from inspect import currentframe, isawaitable from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeAlias, TypedDict, cast if sys.version_info >= (3, 13): @@ -153,18 +153,30 @@ def _mcp_tool_execution_exception( class MCPSpecificApproval(TypedDict, total=False): """Represents the specific approval mode for an MCP tool. - When using this mode, the user must specify which tools always or never require approval. + When using this mode, the user specifies which tools always or never require approval and may set + ``default`` for every other tool and prompt. Names match raw remote names or unambiguous prefixed names when the raw name already matches its normalized form. Normalized-only aliases do not match. Discovery raises ``ToolExecutionException`` if a configured name matches multiple raw remote names. + A name list given as a single string is treated as one name. + + A tool or prompt resolves in this order: a name in ``always_require_approval`` requires approval, otherwise + a name in ``never_require_approval`` does not, otherwise ``default`` applies. Without ``default``, a tool or + prompt named in neither list does not require approval, which is the framework default. Setting + ``never_require_approval`` without ``default`` emits a ``FutureWarning`` because this fallback may change + in a future release; set ``default`` explicitly. Hosted MCP tools do not read ``default``. Attributes: always_require_approval: A sequence of tool names that always require approval. never_require_approval: A sequence of tool names that never require approval. + default: The approval mode for tools and prompts named in neither list. Use ``"always_require"`` to + require approval for every tool except those in ``never_require_approval``, or ``"never_require"`` + to require approval only for the tools in ``always_require_approval``. ``None`` means not set. """ always_require_approval: Collection[str] | None never_require_approval: Collection[str] | None + default: Literal["always_require", "never_require"] | None MCPToolResultContentMode = Literal[ @@ -863,6 +875,63 @@ def _mcp_config_candidate_names(*, local_name: str, normalized_name: str, remote return tuple(names) +_MCP_APPROVAL_MODES = ("always_require", "never_require") +_MCP_APPROVAL_DEFAULT_WARNING = ( + "approval_mode sets `never_require_approval` without `default`. Tools and prompts named in neither list use " + '"never_require" today, and this fallback may change in a future release. Set `default` to "never_require" ' + 'to keep the current behavior or to "always_require" to require approval for them.' +) + + +def _mcp_approval_names(names: Collection[str] | None) -> Collection[str]: + """Return approval-policy names, treating a bare string as one name rather than a substring pattern.""" + if names is None: + return () + if isinstance(names, str): + return (names,) + return names + + +def _mcp_approval_default( + approval_mode: MCPSpecificApproval, + *, + warn_stacklevel: int | None = None, +) -> Literal["always_require", "never_require"] | None: + """Return the validated ``default`` of an approval policy, or ``None`` when it is not set. + + The per-tool resolver calls this for every discovered tool and prompt, so it stays silent unless asked. + Constructors pass ``warn_stacklevel``, counted as for ``warnings.warn`` from the caller of this function, to warn + once, at the code that built the tool, when ``never_require_approval`` is used without ``default``. + """ + default = approval_mode.get("default") + if default is None: + if warn_stacklevel is not None and approval_mode.get("never_require_approval") is not None: + warnings.warn(_MCP_APPROVAL_DEFAULT_WARNING, FutureWarning, stacklevel=warn_stacklevel + 1) + return None + if default not in _MCP_APPROVAL_MODES: + raise ValueError(f"approval_mode `default` must be 'always_require' or 'never_require', got {default!r}.") + return default + + +def _mcp_constructor_stacklevel(instance: object) -> int: + """Return the ``warnings.warn`` stacklevel, counted from ``MCPTool.__init__``, of the constructing code. + + Subclass constructors that forward to ``MCPTool.__init__`` are skipped so the warning points at the caller. + """ + stacklevel = 2 + frame = currentframe() + caller: Any = None + try: + # frame is this helper, frame.f_back is MCPTool.__init__, and its f_back is the first constructor caller. + caller = frame.f_back.f_back if frame is not None and frame.f_back is not None else None + while caller is not None and caller.f_code.co_name == "__init__" and caller.f_locals.get("self") is instance: + stacklevel += 1 + caller = caller.f_back + return stacklevel + finally: + del frame, caller + + def _make_mcp_tool_caller( mcp_tool: MCPTool, remote_tool_name: str ) -> Callable[..., Coroutine[Any, Any, str | list[Content]]]: @@ -1151,6 +1220,8 @@ def __init__( raise ValueError( f"tool_result_content must be one of {sorted(allowed_modes)}, got {tool_result_content!r}." ) + if isinstance(approval_mode, dict): + _mcp_approval_default(approval_mode, warn_stacklevel=_mcp_constructor_stacklevel(self)) self.name = name self.description = description or "" self.approval_mode = approval_mode @@ -1596,8 +1667,8 @@ def _validate_config_names(self, functions: Sequence[FunctionTool]) -> None: """Reject configured names that identify more than one raw remote name.""" configured_names = set(self.allowed_tools or ()) if isinstance(self.approval_mode, dict): - configured_names.update(self.approval_mode.get("always_require_approval") or ()) - configured_names.update(self.approval_mode.get("never_require_approval") or ()) + configured_names.update(_mcp_approval_names(self.approval_mode.get("always_require_approval"))) + configured_names.update(_mcp_approval_names(self.approval_mode.get("never_require_approval"))) if not configured_names: return @@ -2864,17 +2935,16 @@ def _determine_approval_mode( self, *candidate_names: str, ) -> Literal["always_require", "never_require"] | None: - if isinstance(self.approval_mode, dict): - if (always_require := self.approval_mode.get("always_require_approval")) and any( - name in always_require for name in candidate_names - ): - return "always_require" - if (never_require := self.approval_mode.get("never_require_approval")) and any( - name in never_require for name in candidate_names - ): - return "never_require" - return None - return self.approval_mode # type: ignore[return-value] + approval_mode = self.approval_mode + if not isinstance(approval_mode, dict): + return approval_mode # type: ignore[return-value] + always_require = _mcp_approval_names(approval_mode.get("always_require_approval")) + never_require = _mcp_approval_names(approval_mode.get("never_require_approval")) + if any(name in always_require for name in candidate_names): + return "always_require" + if any(name in never_require for name in candidate_names): + return "never_require" + return _mcp_approval_default(approval_mode) async def _load_configured_discovery_locked(self) -> None: """Load configured session discovery while the caller holds the discovery lock.""" @@ -3735,9 +3805,12 @@ def __init__( approval_mode: The approval mode for the tool. This can be: - "always_require": The tool always requires approval before use. - "never_require": The tool never requires approval before use. - - A dict with keys `always_require_approval` or `never_require_approval`, - followed by a sequence of strings with the names of the relevant tools. - A tool should not be listed in both, if so, it will require approval. + - A dict with keys `always_require_approval` and `never_require_approval`, each a sequence of + strings with the names of the relevant tools, and an optional `default` ("always_require" or + "never_require") for tools and prompts named in neither list. Without `default`, those tools + and prompts do not require approval, and setting `never_require_approval` without `default` + emits a ``FutureWarning``. + A tool listed in both lists requires approval. ``None`` (the default) does not require approval. allowed_tools: Optional allow-list of MCP tool names to expose as functions. ``None`` (the default) exposes every tool advertised by the MCP server. A non-empty collection exposes only the tools whose names appear in it. @@ -3992,9 +4065,12 @@ def __init__( approval_mode: The approval mode for the tool. This can be: - "always_require": The tool always requires approval before use. - "never_require": The tool never requires approval before use. - - A dict with keys `always_require_approval` or `never_require_approval`, - followed by a sequence of strings with the names of the relevant tools. - A tool should not be listed in both, if so, it will require approval. + - A dict with keys `always_require_approval` and `never_require_approval`, each a sequence of + strings with the names of the relevant tools, and an optional `default` ("always_require" or + "never_require") for tools and prompts named in neither list. Without `default`, those tools + and prompts do not require approval, and setting `never_require_approval` without `default` + emits a ``FutureWarning``. + A tool listed in both lists requires approval. ``None`` (the default) does not require approval. allowed_tools: Optional allow-list of MCP tool names to expose as functions. ``None`` (the default) exposes every tool advertised by the MCP server. A non-empty collection exposes only the tools whose names appear in it. @@ -4640,9 +4716,12 @@ def __init__( approval_mode: The approval mode for the tool. This can be: - "always_require": The tool always requires approval before use. - "never_require": The tool never requires approval before use. - - A dict with keys `always_require_approval` or `never_require_approval`, - followed by a sequence of strings with the names of the relevant tools. - A tool should not be listed in both, if so, it will require approval. + - A dict with keys `always_require_approval` and `never_require_approval`, each a sequence of + strings with the names of the relevant tools, and an optional `default` ("always_require" or + "never_require") for tools and prompts named in neither list. Without `default`, those tools + and prompts do not require approval, and setting `never_require_approval` without `default` + emits a ``FutureWarning``. + A tool listed in both lists requires approval. ``None`` (the default) does not require approval. allowed_tools: Optional allow-list of MCP tool names to expose as functions. ``None`` (the default) exposes every tool advertised by the MCP server. A non-empty collection exposes only the tools whose names appear in it. diff --git a/python/packages/core/tests/core/mcp_approval_helpers.py b/python/packages/core/tests/core/mcp_approval_helpers.py new file mode 100644 index 00000000000..f741faf20d5 --- /dev/null +++ b/python/packages/core/tests/core/mcp_approval_helpers.py @@ -0,0 +1,58 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Helpers for the MCP approval-policy tests.""" + +from collections.abc import Sequence +from typing import Any + +from agent_framework import ( + ChatResponse, + ChatResponseUpdate, + Content, + FunctionTool, + MCPStreamableHTTPTool, + Message, +) + + +class ForwardingMCPTool(MCPStreamableHTTPTool): + """Transport subclass whose constructor forwards to the transport constructor.""" + + def __init__(self, **kwargs: Any) -> None: + super().__init__(name="docs", url="https://example.com/mcp", **kwargs) + + +async def run_function_call( + client: Any, + tools: Sequence[FunctionTool], + function_name: str, + *, + streaming: bool, +) -> ChatResponse: + """Run one scripted model turn that calls ``function_name`` through the function-invocation loop.""" + function_call = Content.from_function_call(call_id=f"call-{function_name}", name=function_name, arguments={}) + messages = [Message(role="user", contents=["run"])] + if streaming: + client.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[function_call], finish_reason="tool_calls")], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")], finish_reason="stop")], + ] + stream = client.get_response(messages, stream=True, options={"tools": list(tools)}) + async for _ in stream: + pass + return await stream.get_final_response() + client.run_responses = [ + ChatResponse(messages=Message(role="assistant", contents=[function_call])), + ChatResponse(messages=Message(role="assistant", contents=["done"])), + ] + return await client.get_response(messages, options={"tools": list(tools)}) + + +def approval_request_names(response: ChatResponse) -> list[str | None]: + """Return the function names of the approval requests in a response.""" + return [ + content.function_call.name + for message in response.messages + for content in message.contents + if content.type == "function_approval_request" and content.function_call is not None + ] diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 3b814aa05d3..0284baeb5ae 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -64,6 +64,8 @@ ) from agent_framework.exceptions import ToolException, ToolExecutionException +from .mcp_approval_helpers import ForwardingMCPTool, approval_request_names, run_function_call + # Integration test skip condition skip_if_mcp_integration_tests_disabled = pytest.mark.skipif( os.getenv("LOCAL_MCP_URL", "") == "", @@ -354,9 +356,10 @@ async def test_load_tools_with_tool_name_prefix_preserves_matching_configuration [ (["docs_search"], None), (None, {"always_require_approval": ["docs_search"]}), - (None, {"never_require_approval": ["docs_search"]}), + (None, {"always_require_approval": "docs_search"}), + (None, {"never_require_approval": ["docs_search"], "default": "never_require"}), ], - ids=["allowed", "always_require", "never_require"], + ids=["allowed", "always_require", "always_require_single_string", "never_require"], ) @pytest.mark.parametrize("remote_names", [("search", "docs_search"), ("docs_search", "search")]) @pytest.mark.parametrize("paginated", [False, True]) @@ -409,7 +412,7 @@ async def list_tools(params: types.PaginatedRequestParams | None = None) -> type [ (["docs_search"], None), (None, {"always_require_approval": ["docs_search"]}), - (None, {"never_require_approval": ["docs_search"]}), + (None, {"never_require_approval": ["docs_search"], "default": "never_require"}), ], ids=["allowed", "always_require", "never_require"], ) @@ -468,7 +471,7 @@ async def list_tools(params: types.PaginatedRequestParams | None = None) -> type [ (["docs_search"], None, "never_require"), (None, {"always_require_approval": ["docs_search"]}, "always_require"), - (None, {"never_require_approval": ["docs_search"]}, "never_require"), + (None, {"never_require_approval": ["docs_search"], "default": "never_require"}, "never_require"), ], ids=["allowed", "always_require", "never_require"], ) @@ -682,6 +685,7 @@ async def test_overlapping_names_accept_unambiguous_policy( approval_mode={ "always_require_approval": ["search"], "never_require_approval": ["search", "docs_docs_search"], + "default": "never_require", }, ) mock_session = AsyncMock() @@ -3125,9 +3129,26 @@ def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]: { "always_require_approval": ["tool_one"], "never_require_approval": ["tool_two"], + "default": "never_require", }, {"tool_one": "always_require", "tool_two": "never_require"}, ), + ( + {"always_require_approval": ["tool_one"]}, + {"tool_one": "always_require", "tool_two": "never_require"}, + ), + ( + {"default": "always_require", "never_require_approval": ["tool_one"]}, + {"tool_one": "never_require", "tool_two": "always_require"}, + ), + ( + {"default": "never_require", "always_require_approval": ["tool_one"]}, + {"tool_one": "always_require", "tool_two": "never_require"}, + ), + ( + {"default": "always_require"}, + {"tool_one": "always_require", "tool_two": "always_require"}, + ), ], ) async def test_mcp_tool_approval_mode(approval_mode, expected_approvals): @@ -3178,15 +3199,259 @@ def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]: def test_mcp_tool_approval_mode_returns_none_for_unmatched_names() -> None: - tool = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] + with pytest.warns(FutureWarning, match="never_require_approval"): + tool = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] + name="test_tool", + approval_mode={ + "always_require_approval": ["tool_one"], + "never_require_approval": ["tool_two"], + }, + ) + + assert tool._determine_approval_mode("tool_three") is None + + +@pytest.mark.parametrize( + ("approval_mode", "expected"), + [ + ({"always_require_approval": ["tool_one"]}, None), + ({}, None), + ({"default": None}, None), + ({"default": "always_require"}, "always_require"), + ({"default": "never_require"}, "never_require"), + ({"default": "always_require", "never_require_approval": ["tool_one"]}, "always_require"), + ({"default": "never_require", "always_require_approval": ["tool_one"]}, "never_require"), + ], + ids=["always_list_only", "empty", "default_none", "default_always", "default_never", "allow_list", "deny_list"], +) +def test_mcp_tool_approval_default_applies_to_names_in_neither_list( + approval_mode: MCPSpecificApproval, expected: str | None +) -> None: + tool = MCPTool(name="test_tool", approval_mode=approval_mode) # type: ignore[abstract] # ty: ignore[call-non-callable] + + assert tool._determine_approval_mode("tool_three") == expected + + +@pytest.mark.parametrize( + ("approval_mode", "expected"), + [ + ({"default": "always_require", "never_require_approval": ["tool_one"]}, "never_require"), + ({"default": "never_require", "always_require_approval": ["tool_one"]}, "always_require"), + ( + { + "default": "never_require", + "always_require_approval": ["tool_one"], + "never_require_approval": ["tool_one"], + }, + "always_require", + ), + ], + ids=["never_list_beats_default", "always_list_beats_default", "always_list_beats_never_list"], +) +def test_mcp_tool_approval_lists_take_precedence_over_default( + approval_mode: MCPSpecificApproval, expected: str +) -> None: + tool = MCPTool(name="test_tool", approval_mode=approval_mode) # type: ignore[abstract] # ty: ignore[call-non-callable] + + assert tool._determine_approval_mode("tool_one") == expected + + +@pytest.mark.parametrize("default", ["sometimes", "always", "", True, 1]) +def test_mcp_tool_rejects_invalid_approval_default(default: Any) -> None: + with pytest.raises(ValueError, match="`default` must be"): + MCPTool(name="test_tool", approval_mode={"default": default}) # type: ignore[abstract] # ty: ignore[call-non-callable] + + +@pytest.mark.parametrize( + "approval_mode", + [ + {"never_require_approval": ["tool_one"]}, + {"never_require_approval": []}, + {"always_require_approval": ["tool_one"], "never_require_approval": ["tool_two"]}, + ], + ids=["never_list", "empty_never_list", "both_lists"], +) +def test_mcp_tool_warns_when_never_list_has_no_default(approval_mode: MCPSpecificApproval) -> None: + with pytest.warns(FutureWarning, match="without `default`") as recorded: + MCPStreamableHTTPTool(name="docs", url="https://example.com/mcp", approval_mode=approval_mode) + + assert [warning.category for warning in recorded] == [FutureWarning] + assert recorded[0].filename == __file__ + + +def test_mcp_tool_approval_warning_points_at_the_constructing_caller() -> None: + approval_mode: MCPSpecificApproval = {"never_require_approval": ["tool_one"]} + with pytest.warns(FutureWarning, match="without `default`") as recorded: + frame = inspect.currentframe() + assert frame is not None + expected_line = frame.f_lineno + 1 + ForwardingMCPTool(approval_mode=approval_mode) + + assert recorded[0].filename == __file__ + assert recorded[0].lineno == expected_line + + +async def test_mcp_tool_approval_warning_is_not_repeated_during_discovery() -> None: + with pytest.warns(FutureWarning, match="without `default`"): + tool = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] + name="docs", + approval_mode={"never_require_approval": ["search_docs"]}, + ) + mock_session = AsyncMock() + tool.session = mock_session + mock_session.list_tools = AsyncMock( + return_value=types.ListToolsResult( + tools=[ + types.Tool(name="search_docs", input_schema={"type": "object"}), + types.Tool(name="delete_records", input_schema={"type": "object"}), + ] + ) + ) + mock_session.list_prompts = AsyncMock( + return_value=types.ListPromptsResult(prompts=[types.Prompt(name="summarize")]) + ) + + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + await tool.load_tools() + await tool.load_prompts() + await tool.message_handler(types.ToolListChangedNotification()) + await asyncio.gather(*tool._pending_reload_tasks) + + assert [warning for warning in recorded if issubclass(warning.category, FutureWarning)] == [] + assert {function.name: function.approval_mode for function in tool.functions} == { + "search_docs": "never_require", + "delete_records": "never_require", + "summarize": "never_require", + } + + +@pytest.mark.parametrize( + "approval_mode", + [ + None, + "always_require", + "never_require", + {}, + {"always_require_approval": ["tool_one"]}, + {"never_require_approval": None}, + {"never_require_approval": ["tool_one"], "default": "never_require"}, + {"never_require_approval": ["tool_one"], "default": "always_require"}, + ], + ids=[ + "none", + "always_string", + "never_string", + "empty_dict", + "always_list", + "never_list_none", + "never_list_with_never_default", + "never_list_with_always_default", + ], +) +def test_mcp_tool_does_not_warn_when_approval_fallback_is_explicit_or_unused(approval_mode: Any) -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error", FutureWarning) + MCPTool(name="test_tool", approval_mode=approval_mode) # type: ignore[abstract] # ty: ignore[call-non-callable] + + +def test_mcp_tool_approval_names_given_as_one_string_match_exact_names_only() -> None: + deny_list = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] name="test_tool", - approval_mode={ - "always_require_approval": ["tool_one"], - "never_require_approval": ["tool_two"], - }, + approval_mode={"always_require_approval": "delete_records", "default": "never_require"}, + ) + allow_list = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] + name="test_tool", + approval_mode={"never_require_approval": "search_docs", "default": "always_require"}, ) - assert tool._determine_approval_mode("tool_three") is None + assert deny_list._determine_approval_mode("delete_records") == "always_require" + assert deny_list._determine_approval_mode("delete") == "never_require" + assert allow_list._determine_approval_mode("search_docs") == "never_require" + assert allow_list._determine_approval_mode("search") == "always_require" + + +async def test_tools_added_after_list_changed_use_approval_default() -> None: + tool = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] + name="docs", + approval_mode={"default": "always_require", "never_require_approval": ["search_docs"]}, + ) + mock_session = AsyncMock() + tool.session = mock_session + search_docs = types.Tool(name="search_docs", input_schema={"type": "object"}) + delete_records = types.Tool(name="delete_records", input_schema={"type": "object"}) + mock_session.list_tools = AsyncMock( + side_effect=[ + types.ListToolsResult(tools=[search_docs]), + types.ListToolsResult(tools=[search_docs, delete_records]), + ] + ) + + await tool.load_tools() + + assert {function.name: function.approval_mode for function in tool.functions} == {"search_docs": "never_require"} + + await tool.message_handler(types.ToolListChangedNotification()) + await asyncio.gather(*tool._pending_reload_tasks) + + assert {function.name: function.approval_mode for function in tool.functions} == { + "search_docs": "never_require", + "delete_records": "always_require", + } + + +async def test_prompts_use_approval_default() -> None: + tool = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] + name="docs", + approval_mode={"default": "always_require", "never_require_approval": ["summarize"]}, + ) + mock_session = AsyncMock() + tool.session = mock_session + mock_session.list_prompts = AsyncMock( + return_value=types.ListPromptsResult(prompts=[types.Prompt(name="summarize"), types.Prompt(name="draft")]) + ) + + await tool.load_prompts() + + assert {function.name: function.approval_mode for function in tool.functions} == { + "summarize": "never_require", + "draft": "always_require", + } + + +@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"]) +async def test_mcp_tool_named_in_neither_list_waits_for_approval_with_default( + chat_client_base: Any, + streaming: bool, +) -> None: + tool = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] + name="docs", + approval_mode={"default": "always_require", "never_require_approval": ["search_docs"]}, + ) + mock_session = AsyncMock() + tool.session = mock_session + mock_session.list_tools = AsyncMock( + return_value=types.ListToolsResult( + tools=[ + types.Tool(name="search_docs", input_schema={"type": "object", "properties": {}}), + types.Tool(name="delete_records", input_schema={"type": "object", "properties": {}}), + ] + ) + ) + mock_session.call_tool = AsyncMock( + return_value=types.CallToolResult(content=[types.TextContent(type="text", text="ok")]) + ) + await tool.load_tools() + + exempt_response = await run_function_call(chat_client_base, tool.functions, "search_docs", streaming=streaming) + + assert approval_request_names(exempt_response) == [] + assert mock_session.call_tool.await_count == 1 + + gated_response = await run_function_call(chat_client_base, tool.functions, "delete_records", streaming=streaming) + + assert approval_request_names(gated_response) == ["delete_records"] + assert mock_session.call_tool.await_count == 1 @pytest.mark.parametrize( @@ -3817,8 +4082,19 @@ def _tool() -> None: assert result[0].text == "MCP tool name 'shared' is ambiguous." -async def test_mcp_progressive_loaded_tool_preserves_remote_approval_mode() -> None: - server = await _load_progressive_test_server(approval_mode={"always_require_approval": ["tool_two"]}) +@pytest.mark.parametrize( + ("approval_mode", "expected"), + [ + ({"always_require_approval": ["tool_two"]}, "always_require"), + ({"default": "always_require", "never_require_approval": ["tool_one"]}, "always_require"), + ({"default": "never_require", "always_require_approval": ["tool_one"]}, "never_require"), + ], + ids=["listed", "default_always", "default_never"], +) +async def test_mcp_progressive_loaded_tool_preserves_remote_approval_mode( + approval_mode: MCPSpecificApproval, expected: str +) -> None: + server = await _load_progressive_test_server(approval_mode=approval_mode) load_tool = server.functions[1] context = FunctionInvocationContext( function=load_tool, @@ -3837,7 +4113,7 @@ async def test_mcp_progressive_loaded_tool_preserves_remote_approval_mode() -> N if tool.name == "tool_two" # type: ignore[union-attr] # ty: ignore[unresolved-attribute] ), ) - assert loaded_tool.approval_mode == "always_require" + assert loaded_tool.approval_mode == expected async def test_mcp_progressive_loaded_http_tool_preserves_runtime_kwargs_for_headers() -> None: