feat: #2807 support callable approval policies for local MCP servers (#2818)

Author: seratch

src/agents/mcp/__init__.py

@@ -1,6 +1,7 @@
 try:
     from .manager import MCPServerManager
     from .server import (
+        LocalMCPApprovalCallable,
         MCPServer,
         MCPServerSse,
         MCPServerSseParams,
@@ -32,6 +33,7 @@
     "MCPServerStreamableHttp",
     "MCPServerStreamableHttpParams",
     "MCPServerManager",
+    "LocalMCPApprovalCallable",
     "MCPUtil",
     "MCPToolMetaContext",
     "MCPToolMetaResolver",

src/agents/mcp/server.py

@@ -63,13 +63,31 @@ class RequireApprovalObject(TypedDict, total=False):
 
 RequireApprovalPolicy = Literal["always", "never"]
 RequireApprovalMapping = dict[str, RequireApprovalPolicy]
+if TYPE_CHECKING:
+    LocalMCPApprovalCallable = Callable[
+        [RunContextWrapper[Any], "AgentBase", MCPTool],
+        MaybeAwaitable[bool],
+    ]
+else:
+    LocalMCPApprovalCallable = Callable[..., Any]
+
 if TYPE_CHECKING:
     RequireApprovalSetting = (
-        RequireApprovalPolicy | RequireApprovalObject | RequireApprovalMapping | bool | None
+        RequireApprovalPolicy
+        | RequireApprovalObject
+        | RequireApprovalMapping
+        | LocalMCPApprovalCallable
+        | bool
+        | None
     )
 else:
     RequireApprovalSetting = Union[  # noqa: UP007
-        RequireApprovalPolicy, RequireApprovalObject, RequireApprovalMapping, bool, None
+        RequireApprovalPolicy,
+        RequireApprovalObject,
+        RequireApprovalMapping,
+        LocalMCPApprovalCallable,
+        bool,
+        None,
     ]
 
 
@@ -220,8 +238,10 @@ def __init__(
                 default will cause duplicate content. You can set this to True if you know the
                 server will not duplicate the structured content in the `tool_result.content`.
             require_approval: Approval policy for tools on this server. Accepts "always"/"never",
-                a dict of tool names to those values, a boolean, or an object with always/never
-                tool lists (mirroring TS requireApproval). Normalized into a needs_approval policy.
+                a dict of tool names to those values, a boolean, an object with always/never
+                tool lists (mirroring TS requireApproval), or a sync/async callable that receives
+                `(run_context, agent, tool)` and returns whether the tool call needs approval.
+                Normalized into a needs_approval policy.
             failure_error_function: Optional function used to convert MCP tool failures into
                 a model-visible error message. If explicitly set to None, tool errors will be
                 raised instead of converted. If left unset, the agent-level configuration (or
@@ -408,6 +428,9 @@ def _is_tool_list_schema(value: object) -> bool:
                     tool_mapping[str(name)] = _to_bool(value)
             return tool_mapping
 
+        if callable(require_approval):
+            return require_approval
+
         if isinstance(require_approval, bool):
             return require_approval
 
@@ -418,7 +441,12 @@ def _get_needs_approval_for_tool(
         tool: MCPTool,
         agent: AgentBase | None,
     ) -> bool | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]]:
-        """Return a FunctionTool.needs_approval value for a given MCP tool."""
+        """Return a FunctionTool.needs_approval value for a given MCP tool.
+
+        Legacy callers may omit ``agent`` when using ``MCPUtil.to_function_tool()`` directly.
+        When approval is configured with a callable policy and no agent is available, this method
+        returns ``True`` to preserve the historical fail-closed behavior.
+        """
 
         policy = self._needs_approval_policy
 

tests/mcp/test_mcp_approval.py

@@ -1,6 +1,9 @@
+import asyncio
+
 import pytest
+from mcp.types import Tool as MCPTool
 
-from agents import Agent, Runner
+from agents import Agent, RunContextWrapper, Runner
 
 from ..fake_model import FakeModel
 from ..test_responses import get_function_tool_call, get_text_message
@@ -122,3 +125,96 @@ async def test_mcp_require_approval_mapping_allows_policy_keyword_tool_names():
 
     second = await Runner.run(agent, "call never")
     assert not second.interruptions, "tool named 'never' should not require approval"
+
+
+@pytest.mark.asyncio
+async def test_mcp_require_approval_callable_can_allow_and_block_by_tool_name():
+    """Callable policies should decide approval dynamically for each MCP tool."""
+
+    seen: list[str] = []
+
+    def require_approval(
+        _run_context: RunContextWrapper[object | None],
+        _agent: Agent,
+        tool: MCPTool,
+    ) -> bool:
+        seen.append(tool.name)
+        return tool.name == "guarded"
+
+    server = FakeMCPServer(require_approval=require_approval)
+    server.add_tool("guarded", {"type": "object", "properties": {}})
+    server.add_tool("safe", {"type": "object", "properties": {}})
+
+    model = FakeModel()
+    agent = Agent(name="TestAgent", model=model, mcp_servers=[server])
+
+    queue_function_call_and_text(
+        model,
+        get_function_tool_call("guarded", "{}"),
+        followup=[get_text_message("guarded done")],
+    )
+    first = await Runner.run(agent, "call guarded")
+    assert first.interruptions, "guarded should require approval via callable policy"
+    assert first.interruptions[0].tool_name == "guarded"
+
+    resumed = await resume_after_first_approval(agent, first, always_approve=True)
+    assert resumed.final_output == "guarded done"
+
+    queue_function_call_and_text(
+        model,
+        get_function_tool_call("safe", "{}"),
+        followup=[get_text_message("safe done")],
+    )
+    second = await Runner.run(agent, "call safe")
+    assert not second.interruptions, "safe should bypass approval via callable policy"
+    assert second.final_output == "safe done"
+
+    assert seen == ["guarded", "guarded", "safe"]
+
+
+@pytest.mark.asyncio
+async def test_mcp_require_approval_async_callable_uses_run_context():
+    """Async callable policies should receive the run context and be awaited."""
+
+    seen_contexts: list[object | None] = []
+
+    async def require_approval(
+        run_context: RunContextWrapper[dict[str, bool] | None],
+        _agent: Agent,
+        _tool,
+    ) -> bool:
+        seen_contexts.append(run_context.context)
+        await asyncio.sleep(0)
+        return bool(run_context.context and run_context.context.get("needs_approval"))
+
+    server = FakeMCPServer(require_approval=require_approval)
+    server.add_tool("conditional", {"type": "object", "properties": {}})
+
+    model = FakeModel()
+    agent = Agent(name="TestAgent", model=model, mcp_servers=[server])
+
+    queue_function_call_and_text(
+        model,
+        get_function_tool_call("conditional", "{}"),
+        followup=[get_text_message("approved path")],
+    )
+    first = await Runner.run(agent, "call conditional", context={"needs_approval": True})
+    assert first.interruptions, "run context should be able to trigger approval"
+
+    resumed = await resume_after_first_approval(agent, first, always_approve=True)
+    assert resumed.final_output == "approved path"
+
+    queue_function_call_and_text(
+        model,
+        get_function_tool_call("conditional", "{}"),
+        followup=[get_text_message("no approval path")],
+    )
+    second = await Runner.run(agent, "call conditional", context={"needs_approval": False})
+    assert not second.interruptions, "run context should be able to skip approval"
+    assert second.final_output == "no approval path"
+
+    assert seen_contexts == [
+        {"needs_approval": True},
+        {"needs_approval": True},
+        {"needs_approval": False},
+    ]

tests/mcp/test_mcp_util.py

@@ -676,6 +676,78 @@ def require_approval(
     assert function_tool.needs_approval is True
 
 
+@pytest.mark.asyncio
+async def test_to_function_tool_callable_policy_uses_agent_and_tool():
+    """Callable require_approval policies should bridge into FunctionTool.needs_approval."""
+
+    captured: dict[str, Any] = {}
+
+    def require_approval(
+        run_context: RunContextWrapper[Any],
+        agent: Agent,
+        tool: MCPTool,
+    ) -> bool:
+        captured["run_context"] = run_context
+        captured["agent"] = agent
+        captured["tool"] = tool
+        return tool.name == "guarded_tool"
+
+    server = FakeMCPServer(require_approval=require_approval)
+    tool = MCPTool(name="guarded_tool", inputSchema={})
+    agent = Agent(name="test-agent")
+
+    function_tool = MCPUtil.to_function_tool(
+        tool,
+        server,
+        convert_schemas_to_strict=False,
+        agent=agent,
+    )
+
+    assert callable(function_tool.needs_approval)
+
+    run_context = RunContextWrapper(context={"request_id": "req_123"})
+    needs_approval = await function_tool.needs_approval(run_context, {}, "call_123")
+
+    assert needs_approval is True
+    assert captured["run_context"] is run_context
+    assert captured["agent"] is agent
+    assert captured["tool"].name == "guarded_tool"
+
+
+@pytest.mark.asyncio
+async def test_to_function_tool_async_callable_policy_is_awaited():
+    """Async require_approval policies should be awaited before tool execution."""
+
+    async def require_approval(
+        _run_context: RunContextWrapper[Any],
+        _agent: Agent,
+        tool: MCPTool,
+    ) -> bool:
+        await asyncio.sleep(0)
+        return tool.name == "async_guarded_tool"
+
+    server = FakeMCPServer(require_approval=require_approval)
+    tool = MCPTool(name="async_guarded_tool", inputSchema={})
+    agent = Agent(name="test-agent")
+
+    function_tool = MCPUtil.to_function_tool(
+        tool,
+        server,
+        convert_schemas_to_strict=False,
+        agent=agent,
+    )
+
+    assert callable(function_tool.needs_approval)
+
+    needs_approval = await function_tool.needs_approval(
+        RunContextWrapper(context=None),
+        {},
+        "call_async_123",
+    )
+
+    assert needs_approval is True
+
+
 @pytest.mark.asyncio
 async def test_mcp_tool_failure_error_function_agent_default():
     """Agent-level failure_error_function should handle MCP tool failures."""