diff --git a/google/genai/_extra_utils.py b/google/genai/_extra_utils.py index 129c05f7d..1c2dddb30 100644 --- a/google/genai/_extra_utils.py +++ b/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 diff --git a/google/genai/_mcp_utils.py b/google/genai/_mcp_utils.py index 6cd632252..d1af29c24 100644 --- a/google/genai/_mcp_utils.py +++ b/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") diff --git a/google/genai/_transformers.py b/google/genai/_transformers.py index dc09a6ec6..1b03b088e 100644 --- a/google/genai/_transformers.py +++ b/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( diff --git a/google/genai/live.py b/google/genai/live.py index 44f0ada1c..fad01d4c6 100644 --- a/google/genai/live.py +++ b/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( diff --git a/google/genai/types.py b/google/genai/types.py index cc9943404..471a49d3c 100644 --- a/google/genai/types.py +++ b/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]