fix: #2603 preserve MCP tool title metadata across tool call items (#2621)

Author: seratch

src/agents/_mcp_tool_metadata.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True)
+class MCPToolMetadata:
+    """Resolved display metadata for an MCP tool."""
+
+    description: str | None = None
+    title: str | None = None
+
+
+def _get_mapping_or_attr(value: Any, key: str) -> Any:
+    if isinstance(value, Mapping):
+        return value.get(key)
+    return getattr(value, key, None)
+
+
+def _get_non_empty_string(value: Any) -> str | None:
+    if isinstance(value, str) and value:
+        return value
+    return None
+
+
+def resolve_mcp_tool_title(tool: Any) -> str | None:
+    """Return the MCP display title, preferring explicit title over annotations.title."""
+    explicit_title = _get_non_empty_string(_get_mapping_or_attr(tool, "title"))
+    if explicit_title is not None:
+        return explicit_title
+
+    annotations = _get_mapping_or_attr(tool, "annotations")
+    return _get_non_empty_string(_get_mapping_or_attr(annotations, "title"))
+
+
+def resolve_mcp_tool_description(tool: Any) -> str | None:
+    """Return the MCP tool description when present."""
+    return _get_non_empty_string(_get_mapping_or_attr(tool, "description"))
+
+
+def resolve_mcp_tool_description_for_model(tool: Any) -> str:
+    """Return the best model-facing description for an MCP tool.
+
+    MCP distinguishes between a long-form description and a short display title.
+    When the description is absent, fall back to the title so local MCP tools do not
+    become blank function definitions for the model.
+    """
+
+    return resolve_mcp_tool_description(tool) or resolve_mcp_tool_title(tool) or ""
+
+
+def extract_mcp_tool_metadata(tool: Any) -> MCPToolMetadata:
+    """Resolve display metadata from an MCP tool-like object."""
+    return MCPToolMetadata(
+        description=resolve_mcp_tool_description(tool),
+        title=resolve_mcp_tool_title(tool),
+    )
+
+
+def collect_mcp_list_tools_metadata(items: Iterable[Any]) -> dict[tuple[str, str], MCPToolMetadata]:
+    """Collect hosted MCP tool metadata from input/output items.
+
+    Accepts raw `mcp_list_tools` payloads, SDK models, or run items whose `raw_item`
+    contains an `mcp_list_tools` payload.
+    """
+
+    metadata_map: dict[tuple[str, str], MCPToolMetadata] = {}
+
+    for item in items:
+        raw_item = _get_mapping_or_attr(item, "raw_item") or item
+        if _get_mapping_or_attr(raw_item, "type") != "mcp_list_tools":
+            continue
+
+        server_label = _get_non_empty_string(_get_mapping_or_attr(raw_item, "server_label"))
+        tools = _get_mapping_or_attr(raw_item, "tools")
+        if server_label is None or not isinstance(tools, list):
+            continue
+
+        for tool in tools:
+            name = _get_non_empty_string(_get_mapping_or_attr(tool, "name"))
+            if name is None:
+                continue
+            metadata_map[(server_label, name)] = extract_mcp_tool_metadata(tool)
+
+    return metadata_map

src/agents/items.py

@@ -355,6 +355,9 @@ class ToolCallItem(RunItemBase[Any]):
     description: str | None = None
     """Optional tool description if known at item creation time."""
 
+    title: str | None = None
+    """Optional short display label if known at item creation time."""
+
 
 ToolCallOutputTypes: TypeAlias = Union[
     FunctionCallOutput,

src/agents/mcp/util.py

@@ -12,6 +12,7 @@
 from typing_extensions import NotRequired, TypedDict
 
 from .. import _debug
+from .._mcp_tool_metadata import resolve_mcp_tool_description_for_model, resolve_mcp_tool_title
 from ..exceptions import AgentsException, ModelBehaviorError, UserError
 
 try:
@@ -272,7 +273,7 @@ def to_function_tool(
 
         function_tool = _build_wrapped_function_tool(
             name=tool.name,
-            description=tool.description or "",
+            description=resolve_mcp_tool_description_for_model(tool),
             params_json_schema=schema,
             invoke_tool_impl=invoke_func_impl,
             on_handled_error=_build_handled_function_tool_error_handler(
@@ -282,6 +283,7 @@ def to_function_tool(
             failure_error_function=effective_failure_error_function,
             strict_json_schema=is_strict,
             needs_approval=needs_approval,
+            mcp_title=resolve_mcp_tool_title(tool),
         )
         return function_tool
 

src/agents/run_internal/run_loop.py

@@ -12,9 +12,11 @@
 from typing import Any, TypeVar, cast
 
 from openai.types.responses import Response, ResponseCompletedEvent, ResponseOutputItemDoneEvent
+from openai.types.responses.response_output_item import McpCall, McpListTools
 from openai.types.responses.response_prompt_param import ResponsePromptParam
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem
 
+from .._mcp_tool_metadata import collect_mcp_list_tools_metadata
 from .._tool_identity import (
     NamedToolLookupKey,
     build_function_tool_lookup_map,
@@ -1171,6 +1173,9 @@ def _tool_search_fingerprint(raw_item: Any) -> str:
     )
     if isinstance(filtered.input, list):
         filtered.input = deduplicate_input_items_preferring_latest(filtered.input)
+    hosted_mcp_tool_metadata = collect_mcp_list_tools_metadata(streamed_result._model_input_items)
+    if isinstance(filtered.input, list):
+        hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata(filtered.input))
     if server_conversation_tracker is not None:
         logger.debug(
             "filtered.input has %s items; ids=%s",
@@ -1295,6 +1300,9 @@ def _tool_search_fingerprint(raw_item: Any) -> str:
                     )
                 )
 
+            elif isinstance(output_item, McpListTools):
+                hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata([output_item]))
+
             elif isinstance(output_item, TOOL_CALL_TYPES):
                 output_call_id: str | None = getattr(
                     output_item, "call_id", getattr(output_item, "id", None)
@@ -1314,13 +1322,23 @@ def _tool_search_fingerprint(raw_item: Any) -> str:
                         tool_map.get(tool_lookup_key) if tool_lookup_key is not None else None
                     )
                     tool_description: str | None = None
-                    if matched_tool is not None:
+                    tool_title: str | None = None
+                    if isinstance(output_item, McpCall):
+                        metadata = hosted_mcp_tool_metadata.get(
+                            (output_item.server_label, output_item.name)
+                        )
+                        if metadata is not None:
+                            tool_description = metadata.description
+                            tool_title = metadata.title
+                    elif matched_tool is not None:
                         tool_description = getattr(matched_tool, "description", None)
+                        tool_title = getattr(matched_tool, "_mcp_title", None)
 
                     tool_item = ToolCallItem(
                         raw_item=cast(ToolCallItemTypes, output_item),
                         agent=agent,
                         description=tool_description,
+                        title=tool_title,
                     )
                     streamed_result._event_queue.put_nowait(
                         RunItemStreamEvent(item=tool_item, name="tool_called")

src/agents/run_internal/turn_resolution.py

@@ -27,6 +27,7 @@
 )
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem
 
+from .._mcp_tool_metadata import collect_mcp_list_tools_metadata
 from .._tool_identity import (
     build_function_tool_lookup_map,
     get_function_tool_lookup_key,
@@ -1271,6 +1272,7 @@ def process_model_response(
     response: ModelResponse,
     output_schema: AgentOutputSchemaBase | None,
     handoffs: list[Handoff],
+    existing_items: Sequence[RunItem] | None = None,
 ) -> ProcessedResponse:
     items: list[RunItem] = []
 
@@ -1295,6 +1297,8 @@ def process_model_response(
         for tool in all_tools
         if isinstance(tool, HostedMCPTool)
     }
+    hosted_mcp_tool_metadata = collect_mcp_list_tools_metadata(existing_items or ())
+    hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata(response.output))
 
     def _dump_output_item(raw_item: Any) -> dict[str, Any]:
         if isinstance(raw_item, dict):
@@ -1506,19 +1510,15 @@ def _dump_output_item(raw_item: Any) -> dict[str, Any]:
         elif isinstance(output, McpListTools):
             items.append(MCPListToolsItem(raw_item=output, agent=agent))
         elif isinstance(output, McpCall):
-            # Look up MCP tool description from the server's cached tools list.
-            # Tool discovery is async I/O, but this function is sync, so this is best-effort and
-            # only works if the server has cached tool metadata (e.g., when cache_tools_list is
-            # enabled).
-            _mcp_description: str | None = None
-            for _server in agent.mcp_servers:
-                if _server.name == output.server_label:
-                    for _tool in _server.cached_tools or []:
-                        if _tool.name == output.name:
-                            _mcp_description = _tool.description
-                            break
-                    break
-            items.append(ToolCallItem(raw_item=output, agent=agent, description=_mcp_description))
+            metadata = hosted_mcp_tool_metadata.get((output.server_label, output.name))
+            items.append(
+                ToolCallItem(
+                    raw_item=output,
+                    agent=agent,
+                    description=metadata.description if metadata is not None else None,
+                    title=metadata.title if metadata is not None else None,
+                )
+            )
             tools_used.append("mcp")
         elif isinstance(output, ImageGenerationCall):
             items.append(ToolCallItem(raw_item=output, agent=agent))
@@ -1652,6 +1652,7 @@ def _dump_output_item(raw_item: Any) -> dict[str, Any]:
                     raw_item=output,
                     agent=agent,
                     description=func_tool.description,
+                    title=func_tool._mcp_title,
                 )
             )
             functions.append(
@@ -1696,6 +1697,7 @@ async def get_single_step_result_from_response(
         response=new_response,
         output_schema=output_schema,
         handoffs=handoffs,
+        existing_items=pre_step_items,
     )
 
     tool_use_tracker.record_processed_response(agent, processed_response)

src/agents/run_state.py

@@ -116,9 +116,9 @@
 # 2. Keep older readable versions in SUPPORTED_SCHEMA_VERSIONS for backward reads.
 # 3. to_json() always emits CURRENT_SCHEMA_VERSION.
 # 4. Forward compatibility is intentionally fail-fast (older SDKs reject newer versions).
-CURRENT_SCHEMA_VERSION = "1.6"
+CURRENT_SCHEMA_VERSION = "1.7"
 SUPPORTED_SCHEMA_VERSIONS = frozenset(
-    {"1.0", "1.1", "1.2", "1.3", "1.4", "1.5", CURRENT_SCHEMA_VERSION}
+    {"1.0", "1.1", "1.2", "1.3", "1.4", "1.5", "1.6", CURRENT_SCHEMA_VERSION}
 )
 
 _FUNCTION_OUTPUT_ADAPTER: TypeAdapter[FunctionCallOutput] = TypeAdapter(FunctionCallOutput)
@@ -759,6 +759,8 @@ def _serialize_item(self, item: RunItem) -> dict[str, Any]:
             result["allow_bare_name_alias"] = True
         if hasattr(item, "description") and item.description is not None:
             result["description"] = item.description
+        if hasattr(item, "title") and item.title is not None:
+            result["title"] = item.title
 
         return result
 
@@ -2470,10 +2472,16 @@ def _resolve_agent_info(
                 # Tool call items can be function calls, shell calls, apply_patch calls,
                 # MCP calls, etc. Check the type field to determine which type to deserialize as
                 raw_item_tool = _deserialize_tool_call_raw_item(normalized_raw_item)
-                # Preserve description if it was stored with the item
+                # Preserve display metadata if it was stored with the item.
                 description = item_data.get("description")
+                title = item_data.get("title")
                 result.append(
-                    ToolCallItem(agent=agent, raw_item=raw_item_tool, description=description)
+                    ToolCallItem(
+                        agent=agent,
+                        raw_item=raw_item_tool,
+                        description=description,
+                        title=title,
+                    )
                 )
 
             elif item_type == "tool_call_output_item":

src/agents/tool.py

@@ -317,6 +317,9 @@ class FunctionTool:
     _tool_namespace_description: str | None = field(default=None, kw_only=True, repr=False)
     """Internal namespace description used when serializing grouped function tools."""
 
+    _mcp_title: str | None = field(default=None, kw_only=True, repr=False)
+    """Internal MCP display title used for ToolCallItem metadata."""
+
     @property
     def qualified_name(self) -> str:
         """Return the public qualified name used to identify this function tool."""
@@ -418,6 +421,7 @@ def _build_wrapped_function_tool(
     timeout_error_function: ToolErrorFunction | None = None,
     defer_loading: bool = False,
     sync_invoker: bool = False,
+    mcp_title: str | None = None,
 ) -> FunctionTool:
     """Create a FunctionTool with copied-tool-aware failure handling bound in one place."""
     on_invoke_tool = with_function_tool_failure_error_handler(
@@ -442,6 +446,7 @@ def _build_wrapped_function_tool(
             timeout_behavior=timeout_behavior,
             timeout_error_function=timeout_error_function,
             defer_loading=defer_loading,
+            _mcp_title=mcp_title,
         ),
         failure_error_function,
     )

tests/mcp/test_mcp_util.py

@@ -1079,3 +1079,33 @@ async def test_multiple_content_items_without_structured():
     assert result[0]["text"] == "First"
     assert result[1]["type"] == "text"
     assert result[1]["text"] == "Second"
+
+
+def test_to_function_tool_preserves_mcp_title_metadata():
+    server = FakeMCPServer()
+    tool = MCPTool(
+        name="search_docs",
+        inputSchema={},
+        description="Search the docs.",
+        title="Search Docs",
+    )
+
+    function_tool = MCPUtil.to_function_tool(tool, server, convert_schemas_to_strict=False)
+
+    assert function_tool.description == "Search the docs."
+    assert function_tool._mcp_title == "Search Docs"
+
+
+def test_to_function_tool_description_falls_back_to_mcp_title():
+    server = FakeMCPServer()
+    tool = MCPTool(
+        name="search_docs",
+        inputSchema={},
+        description=None,
+        title="Search Docs",
+    )
+
+    function_tool = MCPUtil.to_function_tool(tool, server, convert_schemas_to_strict=False)
+
+    assert function_tool.description == "Search Docs"
+    assert function_tool._mcp_title == "Search Docs"