Python: [Breaking] Add MCP tool approval callback for Azure AI Agent (#14210)

## Summary
- Add an Azure AI Agent callback for MCP tool approval requests.
- Route MCP tool approval submissions through a shared helper in thread
actions.
- Add unit coverage for approval, denial, callback failures, and
per-call decisions.

## Testing
- python -m pytest tests/unit/agents/azure_ai_agent

---------

Copilot-Session: 3e190979-228b-4a82-8e63-5ad14bd4879f

python/samples/concepts/agents/azure_ai_agent/azure_ai_agent_mcp_streaming.py

@@ -40,6 +40,8 @@ async def main() -> None:
 
         # Optionally you may configure to require approval
         # Allowed values are "never" or "always"
+        # When set to "always", Semantic Kernel asks the `mcp_tool_approval_callback` configured on the
+        # agent before each MCP tool call, and denies the call if no callback is configured.
         mcp_tool.set_approval_mode("never")
 
         # 2. Create an agent with the MCP tool on the Azure AI agent service

python/semantic_kernel/agents/__init__.py

@@ -15,6 +15,8 @@ _AGENTS = {
     "AzureAIAgent": ".azure_ai.azure_ai_agent",
     "AzureAIAgentSettings": ".azure_ai.azure_ai_agent_settings",
     "AzureAIAgentThread": ".azure_ai.azure_ai_agent",
+    "MCPToolApprovalCallback": ".azure_ai.mcp_tool_approval",
+    "MCPToolApprovalRequest": ".azure_ai.mcp_tool_approval",
     "AzureAssistantAgent": ".open_ai.azure_assistant_agent",
     "AssistantAgentThread": ".open_ai.openai_assistant_agent",
     "AzureResponsesAgent": ".open_ai.azure_responses_agent",

python/semantic_kernel/agents/azure_ai/agent_thread_actions.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import asyncio
+import inspect
 import logging
 from collections.abc import AsyncIterable
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeVar, cast
@@ -64,6 +65,7 @@ from semantic_kernel.agents.azure_ai.agent_content_generation import (
     get_function_call_contents,
 )
 from semantic_kernel.agents.azure_ai.azure_ai_agent_utils import AzureAIAgentUtils
+from semantic_kernel.agents.azure_ai.mcp_tool_approval import MCPToolApprovalRequest
 from semantic_kernel.agents.open_ai.assistant_content_generation import merge_streaming_function_results
 from semantic_kernel.agents.open_ai.function_action_result import FunctionActionResult
 from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions
@@ -280,16 +282,12 @@ class AgentThreadActions:
                         yield False, generate_mcp_call_content(agent_name=agent.name, mcp_tool_calls=mcp_tool_calls)
 
                         # Create tool approvals for MCP calls
-                        tool_approvals = []
-                        for mcp_call in mcp_tool_calls:
-                            tool_approvals.append(
-                                ToolApproval(
-                                    tool_call_id=mcp_call.id,
-                                    # TODO(evmattso): we don't support manual tool calling yet
-                                    # so we always approve
-                                    approve=True,
-                                )
-                            )
+                        tool_approvals = await cls._build_mcp_tool_approvals(
+                            agent=agent,
+                            thread_id=thread_id,
+                            run_id=run.id,
+                            mcp_tool_calls=mcp_tool_calls,
+                        )
 
                         await agent.client.agents.runs.submit_tool_outputs(
                             run_id=run.id,
@@ -757,17 +755,12 @@ class AgentThreadActions:
                                     output_messages.append(content)
 
                             # Create tool approvals for MCP calls
-                            tool_approvals = []
-                            for mcp_call in mcp_tool_calls:
-                                tool_approvals.append(
-                                    ToolApproval(
-                                        tool_call_id=mcp_call.id,
-                                        approve=True,
-                                        # Note: headers would need to be provided by the MCP tool configuration
-                                        # This is a simplified implementation
-                                        headers={},
-                                    )
-                                )
+                            tool_approvals = await cls._build_mcp_tool_approvals(
+                                agent=agent,
+                                thread_id=thread_id,
+                                run_id=run.id,
+                                mcp_tool_calls=mcp_tool_calls,
+                            )
 
                             handler: BaseAsyncAgentEventHandler = AsyncAgentEventHandler()  # type: ignore
                             await agent.client.agents.runs.submit_tool_outputs_stream(
@@ -1177,6 +1170,71 @@ class AgentThreadActions:
             ],
         )
 
+    @classmethod
+    async def _build_mcp_tool_approvals(
+        cls: type[_T],
+        agent: "AzureAIAgent",
+        thread_id: str,
+        run_id: str,
+        mcp_tool_calls: list[RequiredMcpToolCall],
+    ) -> list[ToolApproval]:
+        """Approve or deny the MCP tool calls the Azure AI Foundry Agent Service is waiting on.
+
+        Each pending call is passed to the agent's `mcp_tool_approval_callback`. When no callback is
+        configured the call is denied, so that a compromised or untrusted MCP server cannot run tools
+        without the developer explicitly opting in.
+        """
+        approvals: list[ToolApproval] = []
+        for mcp_call in mcp_tool_calls:
+            approved = await cls._resolve_mcp_tool_approval(
+                agent,
+                MCPToolApprovalRequest(
+                    agent_name=agent.name,
+                    thread_id=thread_id,
+                    run_id=run_id,
+                    tool_call_id=mcp_call.id,
+                    server_label=mcp_call.server_label,
+                    function_name=mcp_call.name,
+                    arguments=mcp_call.arguments,
+                ),
+            )
+            logger.debug(
+                f"MCP tool call `{mcp_call.name}` from server `{mcp_call.server_label}` was "
+                f"{'approved' if approved else 'denied'} for agent `{agent.name}` and thread `{thread_id}`"
+            )
+            approvals.append(ToolApproval(tool_call_id=mcp_call.id, approve=approved, headers={}))
+
+        return approvals
+
+    @classmethod
+    async def _resolve_mcp_tool_approval(cls: type[_T], agent: "AzureAIAgent", request: MCPToolApprovalRequest) -> bool:
+        """Resolve the approval decision for a single MCP tool call.
+
+        The agent's callback may be synchronous or asynchronous and returns True to approve the call.
+        When no callback is configured, or when the callback fails, the call is denied so that an
+        untrusted MCP server cannot execute tools without the developer opting in.
+        """
+        if agent.mcp_tool_approval_callback is None:
+            logger.warning(
+                f"MCP tool call `{request.function_name}` from server `{request.server_label}` was denied because "
+                "no `mcp_tool_approval_callback` was configured on the agent. Provide a callback to approve or "
+                "deny MCP tool calls."
+            )
+            return False
+
+        try:
+            result = agent.mcp_tool_approval_callback(request)
+            if inspect.isawaitable(result):
+                result = await result
+        except Exception:
+            logger.exception(
+                f"MCP tool approval callback raised an error for tool call `{request.function_name}` from server "
+                f"`{request.server_label}`. Denying the call.",
+            )
+            return False
+
+        return result is True
+
     @classmethod
     def _format_tool_outputs(
         cls: type[_T], fccs: list["FunctionCallContent"], chat_history: "ChatHistory"

python/semantic_kernel/agents/azure_ai/azure_ai_agent.py

@@ -38,6 +38,7 @@ from semantic_kernel.agents import (
 )
 from semantic_kernel.agents.azure_ai.agent_thread_actions import AgentThreadActions
 from semantic_kernel.agents.azure_ai.azure_ai_channel import AzureAIChannel
+from semantic_kernel.agents.azure_ai.mcp_tool_approval import MCPToolApprovalCallback
 from semantic_kernel.agents.channels.agent_channel import AgentChannel
 from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions
 from semantic_kernel.connectors.ai.function_calling_utils import kernel_function_metadata_to_function_call_format
@@ -355,6 +356,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent):
     client: AIProjectClient
     definition: AzureAIAgentModel
     polling_options: RunPollingOptions = Field(default_factory=RunPollingOptions)
+    mcp_tool_approval_callback: MCPToolApprovalCallback | None = None
 
     channel_type: ClassVar[type[AgentChannel]] = AzureAIChannel
 
@@ -368,6 +370,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent):
         plugins: list[KernelPlugin | object] | dict[str, KernelPlugin | object] | None = None,
         polling_options: RunPollingOptions | None = None,
         prompt_template_config: "PromptTemplateConfig | None" = None,
+        mcp_tool_approval_callback: MCPToolApprovalCallback | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the Azure AI Agent.
@@ -384,6 +387,10 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent):
             polling_options: The polling options for the agent.
             prompt_template_config: The prompt template configuration. If this is provided along with
                 instructions, the prompt template will be used in place of the instructions.
+            mcp_tool_approval_callback: A callback invoked for every MCP tool call that the Azure AI Foundry
+                Agent Service asks the caller to approve. It receives an `MCPToolApprovalRequest` and returns
+                True to approve the call, either synchronously or asynchronously. When no callback is
+                provided, such MCP tool calls are denied.
             **kwargs: Additional keyword arguments
         """
         args: dict[str, Any] = {
@@ -423,6 +430,8 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent):
                 args["instructions"] = prompt_template_config.template
         if polling_options is not None:
             args["polling_options"] = polling_options
+        if mcp_tool_approval_callback is not None:
+            args["mcp_tool_approval_callback"] = mcp_tool_approval_callback
         if kwargs:
             args.update(kwargs)
 

python/semantic_kernel/agents/azure_ai/mcp_tool_approval.py

@@ -0,0 +1,35 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Union
+
+from semantic_kernel.utils.feature_stage_decorator import experimental
+
+__all__ = [
+    "MCPToolApprovalCallback",
+    "MCPToolApprovalRequest",
+]
+
+
+@experimental
+@dataclass
+class MCPToolApprovalRequest:
+    """A request to approve or deny a single MCP tool call.
+
+    The Azure AI Foundry Agent Service pauses a run and asks the caller to approve
+    MCP tool calls when the MCP tool is registered with `require_approval="always"`.
+    One instance of this class is passed to the configured approval callback per
+    pending tool call.
+    """
+
+    agent_name: str
+    thread_id: str
+    run_id: str
+    tool_call_id: str
+    server_label: str
+    function_name: str
+    arguments: str
+
+
+MCPToolApprovalCallback = Callable[[MCPToolApprovalRequest], Union[bool, Awaitable[bool]]]

python/tests/unit/agents/azure_ai_agent/test_mcp_tool_approval.py

@@ -0,0 +1,268 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import logging
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from azure.ai.agents.models import (
+    RequiredMcpToolCall,
+    SubmitToolApprovalAction,
+    SubmitToolApprovalDetails,
+    ThreadRun,
+)
+
+from semantic_kernel.agents.azure_ai.agent_thread_actions import AgentThreadActions
+from semantic_kernel.agents.azure_ai.azure_ai_agent import AzureAIAgent
+from semantic_kernel.agents.azure_ai.mcp_tool_approval import (
+    MCPToolApprovalRequest,
+)
+from semantic_kernel.kernel import Kernel
+
+
+def _approve_all(request: MCPToolApprovalRequest) -> bool:
+    return True
+
+
+def _mcp_call(call_id: str = "call-1", name: str = "delete_everything") -> MagicMock:
+    call = MagicMock(spec=RequiredMcpToolCall)
+    call.id = call_id
+    call.name = name
+    call.arguments = '{"target": "prod"}'
+    call.server_label = "attacker_server"
+    return call
+
+
+def _request() -> MCPToolApprovalRequest:
+    return MCPToolApprovalRequest(
+        agent_name="agentName",
+        thread_id="thread-1",
+        run_id="run-1",
+        tool_call_id="call-1",
+        server_label="attacker_server",
+        function_name="delete_everything",
+        arguments="{}",
+    )
+
+
+# region _resolve_mcp_tool_approval
+
+
+def _agent(ai_project_client, ai_agent_definition, callback=None) -> AzureAIAgent:
+    return AzureAIAgent(
+        client=ai_project_client,
+        definition=ai_agent_definition,
+        mcp_tool_approval_callback=callback,
+    )
+
+
+async def test_resolve_denies_when_no_callback_configured(caplog, ai_project_client, ai_agent_definition):
+    agent = _agent(ai_project_client, ai_agent_definition)
+
+    with caplog.at_level(logging.WARNING):
+        approved = await AgentThreadActions._resolve_mcp_tool_approval(agent, _request())
+
+    assert approved is False
+    assert "mcp_tool_approval_callback" in caplog.text
+
+
+async def test_resolve_uses_sync_callback(ai_project_client, ai_agent_definition):
+    def callback(request: MCPToolApprovalRequest) -> bool:
+        assert request.function_name == "delete_everything"
+        return True
+
+    agent = _agent(ai_project_client, ai_agent_definition, callback)
+
+    assert await AgentThreadActions._resolve_mcp_tool_approval(agent, _request()) is True
+
+
+async def test_resolve_uses_async_callback(ai_project_client, ai_agent_definition):
+    async def callback(request: MCPToolApprovalRequest) -> bool:
+        return False
+
+    agent = _agent(ai_project_client, ai_agent_definition, callback)
+
+    assert await AgentThreadActions._resolve_mcp_tool_approval(agent, _request()) is False
+
+
+@pytest.mark.parametrize("value", [True, False])
+async def test_resolve_returns_callback_decision(value: bool, ai_project_client, ai_agent_definition):
+    agent = _agent(ai_project_client, ai_agent_definition, lambda request: value)
+
+    assert await AgentThreadActions._resolve_mcp_tool_approval(agent, _request()) is value
+
+
+async def test_resolve_denies_when_callback_raises(ai_project_client, ai_agent_definition):
+    def callback(request: MCPToolApprovalRequest) -> bool:
+        raise RuntimeError("boom")
+
+    agent = _agent(ai_project_client, ai_agent_definition, callback)
+
+    assert await AgentThreadActions._resolve_mcp_tool_approval(agent, _request()) is False
+
+
+async def test_resolve_denies_on_non_true_return_value(ai_project_client, ai_agent_definition):
+    agent = _agent(ai_project_client, ai_agent_definition, lambda request: "yes please")
+    assert await AgentThreadActions._resolve_mcp_tool_approval(agent, _request()) is False
+
+    agent = _agent(ai_project_client, ai_agent_definition, lambda request: None)
+    assert await AgentThreadActions._resolve_mcp_tool_approval(agent, _request()) is False
+
+
+# region _build_mcp_tool_approvals
+
+
+async def test_build_approvals_denies_by_default(ai_project_client, ai_agent_definition):
+    agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition)
+
+    approvals = await AgentThreadActions._build_mcp_tool_approvals(
+        agent=agent,
+        thread_id="thread-1",
+        run_id="run-1",
+        mcp_tool_calls=[_mcp_call()],
+    )
+
+    assert len(approvals) == 1
+    assert approvals[0].approve is False
+    assert approvals[0].tool_call_id == "call-1"
+
+
+async def test_build_approvals_uses_agent_callback(ai_project_client, ai_agent_definition):
+    seen: list[MCPToolApprovalRequest] = []
+
+    def callback(request: MCPToolApprovalRequest) -> bool:
+        seen.append(request)
+        return True
+
+    agent = AzureAIAgent(
+        client=ai_project_client,
+        definition=ai_agent_definition,
+        mcp_tool_approval_callback=callback,
+    )
+
+    approvals = await AgentThreadActions._build_mcp_tool_approvals(
+        agent=agent,
+        thread_id="thread-1",
+        run_id="run-1",
+        mcp_tool_calls=[_mcp_call()],
+    )
+
+    assert approvals[0].approve is True
+    assert seen[0].agent_name == "agentName"
+    assert seen[0].thread_id == "thread-1"
+    assert seen[0].run_id == "run-1"
+    assert seen[0].server_label == "attacker_server"
+    assert seen[0].function_name == "delete_everything"
+    assert seen[0].arguments == '{"target": "prod"}'
+
+
+async def test_build_approvals_denies_when_callback_returns_false(ai_project_client, ai_agent_definition):
+    agent = AzureAIAgent(
+        client=ai_project_client,
+        definition=ai_agent_definition,
+        mcp_tool_approval_callback=lambda request: False,
+    )
+
+    approvals = await AgentThreadActions._build_mcp_tool_approvals(
+        agent=agent,
+        thread_id="thread-1",
+        run_id="run-1",
+        mcp_tool_calls=[_mcp_call()],
+    )
+
+    assert approvals[0].approve is False
+
+
+async def test_build_approvals_decides_per_call(ai_project_client, ai_agent_definition):
+    def callback(request: MCPToolApprovalRequest) -> bool:
+        return request.function_name == "safe_tool"
+
+    agent = AzureAIAgent(
+        client=ai_project_client,
+        definition=ai_agent_definition,
+        mcp_tool_approval_callback=callback,
+    )
+
+    approvals = await AgentThreadActions._build_mcp_tool_approvals(
+        agent=agent,
+        thread_id="thread-1",
+        run_id="run-1",
+        mcp_tool_calls=[_mcp_call("call-1", "safe_tool"), _mcp_call("call-2", "delete_everything")],
+    )
+
+    assert [(a.tool_call_id, a.approve) for a in approvals] == [("call-1", True), ("call-2", False)]
+
+
+# region end-to-end invoke
+
+
+async def _run_invoke_with_pending_mcp_call(agent) -> AsyncMock:
+    """Drive AgentThreadActions.invoke through a single SubmitToolApprovalAction and return the submit mock."""
+    run = ThreadRun(
+        id="run123",
+        thread_id="thread123",
+        status="running",
+        instructions="test agent",
+        created_at=int(datetime.now(timezone.utc).timestamp()),
+        model="model",
+    )
+
+    agent.client.agents = MagicMock()
+    agent.client.agents.runs = MagicMock()
+    agent.client.agents.runs.create = AsyncMock(return_value=run)
+    agent.client.agents.runs.submit_tool_outputs = AsyncMock()
+
+    async def mock_list_run_steps(*args, **kwargs):
+        return
+        yield  # pragma: no cover - makes this an async generator
+
+    agent.client.agents.run_steps = MagicMock()
+    agent.client.agents.run_steps.list = mock_list_run_steps
+
+    poll_count = 0
+
+    async def mock_poll_run_status(*args, **kwargs):
+        nonlocal poll_count
+        if poll_count == 0:
+            run.status = "requires_action"
+            run.required_action = SubmitToolApprovalAction(
+                submit_tool_approval=SubmitToolApprovalDetails(tool_calls=[_mcp_call()])
+            )
+        else:
+            run.status = "completed"
+        poll_count += 1
+        return run
+
+    with patch.object(AgentThreadActions, "_poll_run_status", side_effect=mock_poll_run_status):
+        async for _ in AgentThreadActions.invoke(
+            agent=agent,
+            thread_id="thread123",
+            kernel=AsyncMock(spec=Kernel),
+        ):
+            pass
+
+    return agent.client.agents.runs.submit_tool_outputs
+
+
+async def test_invoke_denies_mcp_call_without_callback(ai_project_client, ai_agent_definition):
+    agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition)
+
+    submit_tool_outputs = await _run_invoke_with_pending_mcp_call(agent)
+
+    submit_tool_outputs.assert_awaited_once()
+    approvals = submit_tool_outputs.await_args.kwargs["tool_approvals"]
+    assert [a.approve for a in approvals] == [False]
+
+
+async def test_invoke_approves_mcp_call_with_callback(ai_project_client, ai_agent_definition):
+    agent = AzureAIAgent(
+        client=ai_project_client,
+        definition=ai_agent_definition,
+        mcp_tool_approval_callback=_approve_all,
+    )
+
+    submit_tool_outputs = await _run_invoke_with_pending_mcp_call(agent)
+
+    submit_tool_outputs.assert_awaited_once()
+    approvals = submit_tool_outputs.await_args.kwargs["tool_approvals"]
+    assert [a.approve for a in approvals] == [True]