feat: #2367 add MCP tool meta resolver support (#2375)

Author: seratch

pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
     "typing-extensions>=4.12.2, <5",
     "requests>=2.0, <3",
     "types-requests>=2.0, <3",
-    "mcp>=1.11.0, <2; python_version >= '3.10'",
+    "mcp>=1.19.0, <2; python_version >= '3.10'",
 ]
 classifiers = [
     "Typing :: Typed",

src/agents/mcp/__init__.py

@@ -13,6 +13,8 @@
     pass
 
 from .util import (
+    MCPToolMetaContext,
+    MCPToolMetaResolver,
     MCPUtil,
     ToolFilter,
     ToolFilterCallable,
@@ -31,6 +33,8 @@
     "MCPServerStreamableHttpParams",
     "MCPServerManager",
     "MCPUtil",
+    "MCPToolMetaContext",
+    "MCPToolMetaResolver",
     "ToolFilter",
     "ToolFilterCallable",
     "ToolFilterContext",

src/agents/mcp/server.py

@@ -28,7 +28,13 @@
 from ..run_context import RunContextWrapper
 from ..tool import ToolErrorFunction
 from ..util._types import MaybeAwaitable
-from .util import HttpClientFactory, ToolFilter, ToolFilterContext, ToolFilterStatic
+from .util import (
+    HttpClientFactory,
+    MCPToolMetaResolver,
+    ToolFilter,
+    ToolFilterContext,
+    ToolFilterStatic,
+)
 
 
 class RequireApprovalToolList(TypedDict, total=False):
@@ -68,6 +74,7 @@ def __init__(
         use_structured_content: bool = False,
         require_approval: RequireApprovalSetting = None,
         failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET,
+        tool_meta_resolver: MCPToolMetaResolver | None = None,
     ):
         """
         Args:
@@ -83,12 +90,15 @@ def __init__(
                 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
                 SDK default) will be used.
+            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for
+                tool calls. It is invoked by the Agents SDK before calling `call_tool`.
         """
         self.use_structured_content = use_structured_content
         self._needs_approval_policy = self._normalize_needs_approval(
             require_approval=require_approval
         )
         self._failure_error_function = failure_error_function
+        self.tool_meta_resolver = tool_meta_resolver
 
     @abc.abstractmethod
     async def connect(self):
@@ -121,7 +131,12 @@ async def list_tools(
         pass
 
     @abc.abstractmethod
-    async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
+    async def call_tool(
+        self,
+        tool_name: str,
+        arguments: dict[str, Any] | None,
+        meta: dict[str, Any] | None = None,
+    ) -> CallToolResult:
         """Invoke a tool on the server."""
         pass
 
@@ -258,6 +273,7 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         require_approval: RequireApprovalSetting = None,
         failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET,
+        tool_meta_resolver: MCPToolMetaResolver | None = None,
     ):
         """
         Args:
@@ -288,11 +304,14 @@ def __init__(
                 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
                 SDK default) will be used.
+            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for
+                tool calls. It is invoked by the Agents SDK before calling `call_tool`.
         """
         super().__init__(
             use_structured_content=use_structured_content,
             require_approval=require_approval,
             failure_error_function=failure_error_function,
+            tool_meta_resolver=tool_meta_resolver,
         )
         self.session: ClientSession | None = None
         self.exit_stack: AsyncExitStack = AsyncExitStack()
@@ -561,15 +580,24 @@ async def list_tools(
                 f"The server may have disconnected."
             ) from e
 
-    async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
+    async def call_tool(
+        self,
+        tool_name: str,
+        arguments: dict[str, Any] | None,
+        meta: dict[str, Any] | None = None,
+    ) -> CallToolResult:
         """Invoke a tool on the server."""
         if not self.session:
             raise UserError("Server not initialized. Make sure you call `connect()` first.")
         session = self.session
         assert session is not None
 
         try:
-            return await self._run_with_retries(lambda: session.call_tool(tool_name, arguments))
+            if meta is None:
+                return await self._run_with_retries(lambda: session.call_tool(tool_name, arguments))
+            return await self._run_with_retries(
+                lambda: session.call_tool(tool_name, arguments, meta=meta)
+            )
         except httpx.HTTPStatusError as e:
             status_code = e.response.status_code
             raise UserError(
@@ -725,6 +753,7 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         require_approval: RequireApprovalSetting = None,
         failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET,
+        tool_meta_resolver: MCPToolMetaResolver | None = None,
     ):
         """Create a new MCP server based on the stdio transport.
 
@@ -760,17 +789,20 @@ def __init__(
                 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
                 SDK default) will be used.
+            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for
+                tool calls. It is invoked by the Agents SDK before calling `call_tool`.
         """
         super().__init__(
-            cache_tools_list,
-            client_session_timeout_seconds,
-            tool_filter,
-            use_structured_content,
-            max_retry_attempts,
-            retry_backoff_seconds_base,
+            cache_tools_list=cache_tools_list,
+            client_session_timeout_seconds=client_session_timeout_seconds,
+            tool_filter=tool_filter,
+            use_structured_content=use_structured_content,
+            max_retry_attempts=max_retry_attempts,
+            retry_backoff_seconds_base=retry_backoff_seconds_base,
             message_handler=message_handler,
             require_approval=require_approval,
             failure_error_function=failure_error_function,
+            tool_meta_resolver=tool_meta_resolver,
         )
 
         self.params = StdioServerParameters(
@@ -837,6 +869,7 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         require_approval: RequireApprovalSetting = None,
         failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET,
+        tool_meta_resolver: MCPToolMetaResolver | None = None,
     ):
         """Create a new MCP server based on the HTTP with SSE transport.
 
@@ -874,17 +907,20 @@ def __init__(
                 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
                 SDK default) will be used.
+            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for
+                tool calls. It is invoked by the Agents SDK before calling `call_tool`.
         """
         super().__init__(
-            cache_tools_list,
-            client_session_timeout_seconds,
-            tool_filter,
-            use_structured_content,
-            max_retry_attempts,
-            retry_backoff_seconds_base,
+            cache_tools_list=cache_tools_list,
+            client_session_timeout_seconds=client_session_timeout_seconds,
+            tool_filter=tool_filter,
+            use_structured_content=use_structured_content,
+            max_retry_attempts=max_retry_attempts,
+            retry_backoff_seconds_base=retry_backoff_seconds_base,
             message_handler=message_handler,
             require_approval=require_approval,
             failure_error_function=failure_error_function,
+            tool_meta_resolver=tool_meta_resolver,
         )
 
         self.params = params
@@ -954,6 +990,7 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         require_approval: RequireApprovalSetting = None,
         failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET,
+        tool_meta_resolver: MCPToolMetaResolver | None = None,
     ):
         """Create a new MCP server based on the Streamable HTTP transport.
 
@@ -992,17 +1029,20 @@ def __init__(
                 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
                 SDK default) will be used.
+            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for
+                tool calls. It is invoked by the Agents SDK before calling `call_tool`.
         """
         super().__init__(
-            cache_tools_list,
-            client_session_timeout_seconds,
-            tool_filter,
-            use_structured_content,
-            max_retry_attempts,
-            retry_backoff_seconds_base,
+            cache_tools_list=cache_tools_list,
+            client_session_timeout_seconds=client_session_timeout_seconds,
+            tool_filter=tool_filter,
+            use_structured_content=use_structured_content,
+            max_retry_attempts=max_retry_attempts,
+            retry_backoff_seconds_base=retry_backoff_seconds_base,
             message_handler=message_handler,
             require_approval=require_approval,
             failure_error_function=failure_error_function,
+            tool_meta_resolver=tool_meta_resolver,
         )
 
         self.params = params

src/agents/mcp/util.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import copy
 import functools
 import inspect
 import json
@@ -104,6 +105,40 @@ class ToolFilterStatic(TypedDict):
 """A tool filter that can be either a function, static configuration, or None (no filtering)."""
 
 
+@dataclass
+class MCPToolMetaContext:
+    """Context information available to MCP tool meta resolver functions."""
+
+    run_context: RunContextWrapper[Any]
+    """The current run context."""
+
+    server_name: str
+    """The name of the MCP server."""
+
+    tool_name: str
+    """The name of the tool being invoked."""
+
+    arguments: dict[str, Any] | None
+    """The parsed tool arguments."""
+
+
+if TYPE_CHECKING:
+    MCPToolMetaResolver = Callable[
+        [MCPToolMetaContext],
+        MaybeAwaitable[dict[str, Any] | None],
+    ]
+else:
+    MCPToolMetaResolver = Callable[..., Any]
+"""A function that produces MCP request metadata for tool calls.
+
+Args:
+    context: Context information about the tool invocation.
+
+Returns:
+    A dict to send as MCP `_meta`, or None to omit metadata.
+"""
+
+
 def create_static_tool_filter(
     allowed_tool_names: list[str] | None = None,
     blocked_tool_names: list[str] | None = None,
@@ -264,9 +299,57 @@ async def invoke_func(ctx: ToolContext[Any], input_json: str) -> ToolOutput:
             needs_approval=server._get_needs_approval_for_tool(tool, agent),
         )
 
+    @staticmethod
+    def _merge_mcp_meta(
+        resolved_meta: dict[str, Any] | None,
+        explicit_meta: dict[str, Any] | None,
+    ) -> dict[str, Any] | None:
+        if resolved_meta is None and explicit_meta is None:
+            return None
+        merged: dict[str, Any] = {}
+        if resolved_meta is not None:
+            merged.update(resolved_meta)
+        if explicit_meta is not None:
+            merged.update(explicit_meta)
+        return merged
+
+    @classmethod
+    async def _resolve_meta(
+        cls,
+        server: MCPServer,
+        context: RunContextWrapper[Any],
+        tool_name: str,
+        arguments: dict[str, Any] | None,
+    ) -> dict[str, Any] | None:
+        meta_resolver = getattr(server, "tool_meta_resolver", None)
+        if meta_resolver is None:
+            return None
+
+        arguments_copy = copy.deepcopy(arguments) if arguments is not None else None
+        resolver_context = MCPToolMetaContext(
+            run_context=context,
+            server_name=server.name,
+            tool_name=tool_name,
+            arguments=arguments_copy,
+        )
+        result = meta_resolver(resolver_context)
+        if inspect.isawaitable(result):
+            result = await result
+        if result is None:
+            return None
+        if not isinstance(result, dict):
+            raise TypeError("MCP meta resolver must return a dict or None.")
+        return result
+
     @classmethod
     async def invoke_mcp_tool(
-        cls, server: MCPServer, tool: MCPTool, context: RunContextWrapper[Any], input_json: str
+        cls,
+        server: MCPServer,
+        tool: MCPTool,
+        context: RunContextWrapper[Any],
+        input_json: str,
+        *,
+        meta: dict[str, Any] | None = None,
     ) -> ToolOutput:
         """Invoke an MCP tool and return the result as a string."""
         try:
@@ -286,7 +369,12 @@ async def invoke_mcp_tool(
             logger.debug(f"Invoking MCP tool {tool.name} with input {input_json}")
 
         try:
-            result = await server.call_tool(tool.name, json_data)
+            resolved_meta = await cls._resolve_meta(server, context, tool.name, json_data)
+            merged_meta = cls._merge_mcp_meta(resolved_meta, meta)
+            if merged_meta is None:
+                result = await server.call_tool(tool.name, json_data)
+            else:
+                result = await server.call_tool(tool.name, json_data, meta=merged_meta)
         except UserError:
             # Re-raise UserError as-is (it already has a good message)
             raise