Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 25 additions & 26 deletions google/genai/_extra_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down
32 changes: 21 additions & 11 deletions google/genai/_mcp_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
"""Utils for working with MCP tools."""

from importlib.metadata import PackageNotFoundError, version
import sys
import typing
from typing import Any

Expand All @@ -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:
Expand All @@ -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")
Expand Down
18 changes: 7 additions & 11 deletions google/genai/_transformers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand Down Expand Up @@ -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(
Expand Down
45 changes: 21 additions & 24 deletions google/genai/live.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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')

Expand Down Expand Up @@ -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(
Expand Down
53 changes: 35 additions & 18 deletions google/genai/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.'
Expand Down Expand Up @@ -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]
Expand Down
Loading