chore: defer mcp SDK import to first MCP usage

PiperOrigin-RevId: 904039259

Author: google-genai-bot

google/genai/_extra_utils.py

@@ -45,12 +45,6 @@
 else:
   McpClientSession: typing.Type = Any
   McpTool: typing.Type = Any
-  try:
-    from mcp import ClientSession as McpClientSession
-    from mcp.types import Tool as McpTool
-  except ImportError:
-    McpClientSession = None
-    McpTool = None
 
 _DEFAULT_MAX_REMOTE_CALLS_AFC = 10
 
@@ -568,27 +562,32 @@ async def parse_config_for_mcp_sessions(
   parsed_config_copy = parsed_config.model_copy(update={'tools': None})
   if parsed_config.tools:
     parsed_config_copy.tools = []
-    for tool in parsed_config.tools:
-      if McpClientSession is not None and isinstance(tool, McpClientSession):
-        mcp_to_genai_tool_adapter = McpToGenAiToolAdapter(
-            tool, await tool.list_tools()
-        )
-        # Extend the config with the MCP session tools converted to GenAI tools.
-        parsed_config_copy.tools.extend(mcp_to_genai_tool_adapter.tools)
-        for genai_tool in mcp_to_genai_tool_adapter.tools:
-          if genai_tool.function_declarations:
-            for function_declaration in genai_tool.function_declarations:
-              if function_declaration.name:
-                if mcp_to_genai_tool_adapters.get(function_declaration.name):
-                  raise ValueError(
-                      f'Tool {function_declaration.name} is already defined for'
-                      ' the request.'
+    if 'mcp' not in sys.modules:
+      # No MCP tools possible if `mcp` isn't loaded; pass through unchanged.
+      parsed_config_copy.tools.extend(parsed_config.tools)
+    else:
+      from mcp import ClientSession as _McpClientSession
+      for tool in parsed_config.tools:
+        if isinstance(tool, _McpClientSession):
+          mcp_to_genai_tool_adapter = McpToGenAiToolAdapter(
+              tool, await tool.list_tools()
+          )
+          # Extend the config with the MCP session tools converted to GenAI tools.
+          parsed_config_copy.tools.extend(mcp_to_genai_tool_adapter.tools)
+          for genai_tool in mcp_to_genai_tool_adapter.tools:
+            if genai_tool.function_declarations:
+              for function_declaration in genai_tool.function_declarations:
+                if function_declaration.name:
+                  if mcp_to_genai_tool_adapters.get(function_declaration.name):
+                    raise ValueError(
+                        f'Tool {function_declaration.name} is already defined for'
+                        ' the request.'
+                    )
+                  mcp_to_genai_tool_adapters[function_declaration.name] = (
+                      mcp_to_genai_tool_adapter
                   )
-                mcp_to_genai_tool_adapters[function_declaration.name] = (
-                    mcp_to_genai_tool_adapter
-                )
-      else:
-        parsed_config_copy.tools.append(tool)
+        else:
+          parsed_config_copy.tools.append(tool)
 
   return parsed_config_copy, mcp_to_genai_tool_adapters
 

google/genai/_mcp_utils.py

@@ -16,6 +16,7 @@
 """Utils for working with MCP tools."""
 
 from importlib.metadata import PackageNotFoundError, version
+import sys
 import typing
 from typing import Any
 
@@ -28,12 +29,16 @@
 else:
   McpClientSession: typing.Type = Any
   McpTool: typing.Type = Any
-  try:
-    from mcp.types import Tool as McpTool
-    from mcp import ClientSession as McpClientSession
-  except ImportError:
-    McpTool = None
-    McpClientSession = None
+
+
+def _is_mcp_loaded() -> bool:
+  """True iff `mcp` is already imported in this process.
+
+  An MCP tool/session can only exist if the user has already imported `mcp`,
+  so we can gate isinstance checks behind this without ever importing `mcp`
+  ourselves at module load.
+  """
+  return "mcp" in sys.modules
 
 
 def mcp_to_gemini_tool(tool: McpTool) -> types.Tool:
@@ -58,27 +63,32 @@ def mcp_to_gemini_tools(tools: list[McpTool]) -> list[types.Tool]:
 
 def has_mcp_tool_usage(tools: types.ToolListUnion) -> bool:
   """Checks whether the list of tools contains any MCP tools or sessions."""
-  if McpClientSession is None:
+  if not _is_mcp_loaded():
     return False
+  from mcp import ClientSession as _McpClientSession
+  from mcp.types import Tool as _McpTool
+
   for tool in tools:
-    if isinstance(tool, McpTool) or isinstance(tool, McpClientSession):
+    if isinstance(tool, _McpTool) or isinstance(tool, _McpClientSession):
       return True
   return False
 
 
 def has_mcp_session_usage(tools: types.ToolListUnion) -> bool:
   """Checks whether the list of tools contains any MCP sessions."""
-  if McpClientSession is None:
+  if not _is_mcp_loaded():
     return False
+  from mcp import ClientSession as _McpClientSession
+
   for tool in tools:
-    if isinstance(tool, McpClientSession):
+    if isinstance(tool, _McpClientSession):
       return True
   return False
 
 
 def set_mcp_usage_header(headers: dict[str, str]) -> None:
   """Sets the MCP version label in the Google API client header."""
-  if McpClientSession is None:
+  if not _is_mcp_loaded():
     return
   try:
     version_label = version("mcp")

google/genai/_transformers.py

@@ -57,12 +57,6 @@
 else:
   McpClientSession: typing.Type = Any
   McpTool: typing.Type = Any
-  try:
-    from mcp import ClientSession as McpClientSession
-    from mcp.types import Tool as McpTool
-  except ImportError:
-    McpClientSession = None
-    McpTool = None
 
 
 metric_name_sdk_api_map = {
@@ -967,12 +961,14 @@ def t_tool(
             )
         ]
     )
-  elif McpTool is not None and is_duck_type_of(origin, McpTool):
-    return mcp_to_gemini_tool(origin)
-  elif isinstance(origin, dict):
+  if 'mcp' in sys.modules:
+    from mcp.types import Tool as _McpTool
+
+    if is_duck_type_of(origin, _McpTool):
+      return mcp_to_gemini_tool(origin)
+  if isinstance(origin, dict):
     return types.Tool.model_validate(origin)
-  else:
-    return origin
+  return origin
 
 
 def t_tools(

google/genai/live.py

@@ -20,6 +20,7 @@
 import contextlib
 import json
 import logging
+import sys
 import typing
 from typing import Any, AsyncIterator, Optional, Sequence, Union, get_args
 import warnings
@@ -59,22 +60,12 @@
 if typing.TYPE_CHECKING:
   from mcp import ClientSession as McpClientSession
   from mcp.types import Tool as McpTool
-  from ._adapters import McpToGenAiToolAdapter
-  from ._mcp_utils import mcp_to_gemini_tool
 else:
   McpClientSession: typing.Type = Any
   McpTool: typing.Type = Any
-  McpToGenAiToolAdapter: typing.Type = Any
-  try:
-    from mcp import ClientSession as McpClientSession
-    from mcp.types import Tool as McpTool
-    from ._adapters import McpToGenAiToolAdapter
-    from ._mcp_utils import mcp_to_gemini_tool
-  except ImportError:
-    McpClientSession = None
-    McpTool = None
-    McpToGenAiToolAdapter = None
-    mcp_to_gemini_tool = None
+
+from ._adapters import McpToGenAiToolAdapter
+from ._mcp_utils import mcp_to_gemini_tool
 
 logger = logging.getLogger('google_genai.live')
 
@@ -1174,17 +1165,23 @@ async def _t_live_connect_config(
   parameter_model_copy = parameter_model.model_copy(update={'tools': None})
   if parameter_model.tools:
     parameter_model_copy.tools = []
-    for tool in parameter_model.tools:
-      if McpClientSession is not None and isinstance(tool, McpClientSession):
-        mcp_to_genai_tool_adapter = McpToGenAiToolAdapter(
-            tool, await tool.list_tools()
-        )
-        # Extend the config with the MCP session tools converted to GenAI tools.
-        parameter_model_copy.tools.extend(mcp_to_genai_tool_adapter.tools)
-      elif McpTool is not None and isinstance(tool, McpTool):
-        parameter_model_copy.tools.append(mcp_to_gemini_tool(tool))
-      else:
-        parameter_model_copy.tools.append(tool)
+    if 'mcp' not in sys.modules:
+      # No MCP tools possible if `mcp` isn't loaded; pass through unchanged.
+      parameter_model_copy.tools.extend(parameter_model.tools)
+    else:
+      from mcp import ClientSession as _McpClientSession
+      from mcp.types import Tool as _McpTool
+      for tool in parameter_model.tools:
+        if isinstance(tool, _McpClientSession):
+          mcp_to_genai_tool_adapter = McpToGenAiToolAdapter(
+              tool, await tool.list_tools()
+          )
+          # Extend the config with the MCP session tools converted to GenAI tools.
+          parameter_model_copy.tools.extend(mcp_to_genai_tool_adapter.tools)
+        elif isinstance(tool, _McpTool):
+          parameter_model_copy.tools.append(mcp_to_gemini_tool(tool))
+        else:
+          parameter_model_copy.tools.append(tool)
 
   if parameter_model_copy.generation_config is not None:
     warnings.warn(

google/genai/types.py

@@ -25,9 +25,9 @@
 import sys
 import types as builtin_types
 import typing
-from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Union, _UnionGenericAlias  # type: ignore
+from typing import Annotated, Any, Callable, Dict, List, Literal, Optional, Sequence, Union, _UnionGenericAlias  # type: ignore
 import pydantic
-from pydantic import ConfigDict, Field, PrivateAttr, model_validator
+from pydantic import ConfigDict, Field, PrivateAttr, WrapValidator, model_validator
 from typing_extensions import Self, TypedDict
 from . import _common
 from ._operations_converters import (
@@ -63,25 +63,18 @@
   except ImportError:
     PIL_Image = None
 
-_is_mcp_imported = False
 if typing.TYPE_CHECKING:
   from mcp import types as mcp_types
   from mcp import ClientSession as McpClientSession
   from mcp.types import CallToolResult as McpCallToolResult
-
-  _is_mcp_imported = True
 else:
   McpClientSession: typing.Type = Any
   McpCallToolResult: typing.Type = Any
-  try:
-    from mcp import types as mcp_types
-    from mcp import ClientSession as McpClientSession
-    from mcp.types import CallToolResult as McpCallToolResult
 
-    _is_mcp_imported = True
-  except ImportError:
-    McpClientSession = None
-    McpCallToolResult = None
+
+def _is_mcp_imported() -> bool:
+  return 'mcp' in sys.modules
+
 
 if typing.TYPE_CHECKING:
   import yaml
@@ -1789,7 +1782,7 @@ class FunctionResponse(_common.BaseModel):
   def from_mcp_response(
       cls, *, name: str, response: McpCallToolResult
   ) -> 'FunctionResponse':
-    if not _is_mcp_imported:
+    if not _is_mcp_imported():
       raise ValueError(
           'MCP response is not supported. Please ensure that the MCP library is'
           ' imported.'
@@ -4888,17 +4881,41 @@ class ToolDict(TypedDict, total=False):
 
 
 ToolOrDict = Union[Tool, ToolDict]
-if _is_mcp_imported:
+
+
+def _validate_tool_list(v: Any, handler: Any) -> Any:
+  """Pass MCP tool/session objects through Pydantic validation untouched.
+
+  `Tool` has all-optional fields, so without this wrapper Pydantic's default
+  Union resolution would coerce any non-Tool item (e.g. an `mcp.ClientSession`)
+  into an empty `Tool()`. This wrapper dispatches dict -> Tool and leaves
+  every other type identity-preserved so MCP routing downstream still works.
+  """
+  if v is None:
+    return None
+  if not isinstance(v, list):
+    return handler(v)
+  out = []
+  for item in v:
+    if isinstance(item, dict):
+      out.append(Tool.model_validate(item))
+    else:
+      out.append(item)
+  return out
+
+
+if typing.TYPE_CHECKING:
   ToolUnion = Union[Tool, Callable[..., Any], mcp_types.Tool, McpClientSession]
   ToolUnionDict = Union[
       ToolDict, Callable[..., Any], mcp_types.Tool, McpClientSession
   ]
+  ToolListUnion = list[ToolUnion]
+  ToolListUnionDict = list[ToolUnionDict]
 else:
   ToolUnion = Union[Tool, Callable[..., Any]]  # type: ignore[misc]
   ToolUnionDict = Union[ToolDict, Callable[..., Any]]  # type: ignore[misc]
-
-ToolListUnion = list[ToolUnion]
-ToolListUnionDict = list[ToolUnionDict]
+  ToolListUnion = Annotated[list, WrapValidator(_validate_tool_list)]
+  ToolListUnionDict = list[ToolUnionDict]
 
 SchemaUnion = Union[
     dict[Any, Any], type, Schema, builtin_types.GenericAlias, VersionedUnionType  # type: ignore[valid-type]