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
232 changes: 232 additions & 0 deletions agentplatform/_genai/_evals_builtin_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,232 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Built-in tool catalog for Gemini Agent evaluation display.

The Gemini Agents API (``GET agents/{id}``) returns each tool as a bare type
discriminator (e.g. ``{"type": "code_execution"}``) with no parameter schema
or description. The authoritative, full-fidelity expansion lives server-side
in ``cloud/ai/platform/evaluation/utils/interaction_converter.py``.

This module is a **display-only duplicate** of that server catalog, kept here
so ``show()`` can render tools with full names, descriptions, and parameter
schemas without a server round-trip.

**If the server catalog changes, this SDK-side copy must be updated to match.**
"""

from typing import Any, Optional

from google.genai import types as genai_types


def _str_schema(description: str) -> genai_types.Schema:
return genai_types.Schema(type="STRING", description=description)


# Maps a built-in Gemini Agent tool type to the concrete FunctionDeclarations
# the agent actually exposes for that type.
#
# Source of truth: interaction_converter.py, _BUILTIN_TOOL_FUNCTION_DECLARATIONS
BUILTIN_TOOL_DECLARATIONS: dict[str, list[genai_types.FunctionDeclaration]] = {
"code_execution": [
genai_types.FunctionDeclaration(
name="run_command",
description=(
"Runs a shell command on the sandbox VM. If the command does"
" not complete within WaitMsBeforeAsync, it is sent to the"
" background and a CommandId is returned for use with"
" command_status."
),
parameters=genai_types.Schema(
type="OBJECT",
properties={
"CommandLine": _str_schema("The shell command to run."),
"Cwd": _str_schema(
"The current working directory for the command."
),
"WaitMsBeforeAsync": genai_types.Schema(
type="INTEGER",
description=(
"Milliseconds to wait for the command to complete"
" before sending it to the background. Default:"
" 10000."
),
),
"SafeToAutoRun": genai_types.Schema(
type="BOOLEAN",
description=(
"Whether the command is safe to auto-run without"
" user approval."
),
),
},
required=["CommandLine", "Cwd"],
),
),
],
"filesystem": [
genai_types.FunctionDeclaration(
name="view_file",
description="Reads the content of a workspace file.",
),
genai_types.FunctionDeclaration(
name="create_file",
description="Writes content to a new or existing file.",
),
genai_types.FunctionDeclaration(
name="edit_file",
description="Replaces a specific block of text in a file.",
),
genai_types.FunctionDeclaration(
name="list_dir",
description="Lists the files in a directory.",
),
genai_types.FunctionDeclaration(
name="delete_file",
description="Removes a file from the workspace.",
),
genai_types.FunctionDeclaration(
name="move_file",
description="Renames or moves a file.",
),
],
}


# Sandbox-environment orchestration tool declarations.
#
# Source of truth: interaction_converter.py, _SANDBOX_FUNCTION_DECLARATIONS
SANDBOX_DECLARATIONS: list[genai_types.FunctionDeclaration] = [
genai_types.FunctionDeclaration(
name="provision_sandbox",
description="Provisions a sandbox environment.",
parameters=genai_types.Schema(
type="OBJECT",
properties={
"display_name": _str_schema(
"The display name of the sandbox environment."
),
"poll_creation_lro": genai_types.Schema(
type="BOOLEAN",
description=(
"Whether to poll the creation long-running operation."
),
),
},
),
),
genai_types.FunctionDeclaration(
name="load_sandbox",
description="Loads a previously provisioned sandbox environment.",
parameters=genai_types.Schema(
type="OBJECT",
properties={
"reasoning_engine_resource_name": _str_schema(
"The resource name of the reasoning engine. Format:"
" projects/{project}/locations/{location}/reasoningEngines/{id}"
),
"display_name": _str_schema(
"The display name of the sandbox environment. Format: any"
" string."
),
},
),
),
]


def agent_tools_to_config_tools(
agent_tools: Optional[list[Any]],
has_environment: bool = False,
) -> Optional[list[genai_types.Tool]]:
"""Maps Gemini Agents API tools to ``genai_types.Tool`` for display.

Expands built-in agent tool types into their concrete function declarations
using ``BUILTIN_TOOL_DECLARATIONS`` (a display-only duplicate of the
server-side catalog in ``interaction_converter.py``).

Mapping rules:
* ``code_execution`` is expanded to ``run_command`` with full parameter
schema.
* ``filesystem`` is expanded to ``view_file``, ``create_file``,
``edit_file``, ``list_dir``, ``delete_file``, ``move_file``.
* ``google_search`` and ``url_context`` are mapped to their typed
``genai_types.Tool`` variant.
* ``mcp_server`` is represented as a named declaration with a
human-readable label.
* Tools carrying explicit ``function_declarations`` are passed through.
* When ``has_environment`` is True, sandbox orchestration tools
(``provision_sandbox``, ``load_sandbox``) are appended.

Args:
agent_tools: The ``tools`` list from a fetched Gemini agent dict.
has_environment: Whether the agent has a sandbox environment configured.

Returns:
A list of ``genai_types.Tool``, or ``None`` if there are no mappable
tools.
"""
if not agent_tools and not has_environment:
return None
tools: list[genai_types.Tool] = []
for tool in agent_tools or []:
if not isinstance(tool, dict):
continue
tool_type = tool.get("type")
remainder = {k: v for k, v in tool.items() if k != "type"}

# Check the built-in catalog first (code_execution, filesystem).
catalog_decls = BUILTIN_TOOL_DECLARATIONS.get(tool_type or "")
if catalog_decls:
tools.append(
genai_types.Tool(function_declarations=list(catalog_decls))
)
elif tool_type == "google_search":
tools.append(genai_types.Tool(google_search=genai_types.GoogleSearch()))
elif tool_type == "url_context":
tools.append(genai_types.Tool(url_context=genai_types.UrlContext()))
elif "function_declarations" in remainder:
# Real function tool with explicit declarations.
tools.append(genai_types.Tool.model_validate(remainder))
elif tool_type == "mcp_server":
label = remainder.get("name") or remainder.get("url")
description = f"MCP server: {label}" if label else "MCP server."
tools.append(
genai_types.Tool(
function_declarations=[
genai_types.FunctionDeclaration(
name="mcp_server", description=description
)
]
)
)
elif tool_type:
# Unknown built-in: show by name so it isn't silently dropped.
tools.append(
genai_types.Tool(
function_declarations=[
genai_types.FunctionDeclaration(name=tool_type)
]
)
)
elif remainder:
tools.append(genai_types.Tool.model_validate(remainder))

if has_environment:
tools.append(
genai_types.Tool(function_declarations=list(SANDBOX_DECLARATIONS))
)

return tools or None
86 changes: 16 additions & 70 deletions agentplatform/_genai/_evals_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
from tqdm import tqdm
from pydantic import ValidationError

from . import _evals_builtin_tools
from . import _evals_constant
from . import _evals_data_converters
from . import _evals_metric_handlers
Expand Down Expand Up @@ -483,6 +484,7 @@ def _get_default_prompt_template(
and eval_item.evaluation_request
and eval_item.evaluation_request.prompt
and eval_item.evaluation_request.prompt.prompt_template_data
and eval_item.evaluation_request.prompt.prompt_template_data.values
):
if (
"prompt"
Expand Down Expand Up @@ -837,48 +839,7 @@ def _merge_text_parts_in_agent_data(
content.parts = merged_parts


def _agent_tools_to_config_tools(
agent_tools: Optional[list[Any]],
) -> Optional[list[genai_types.Tool]]:
"""Maps Gemini Agents API tools to ``genai_types.Tool`` for an AgentConfig.

The Gemini Agents API returns built-in tool variants (``google_search``,
``code_execution``, ``url_context``) whose schema differs from
``genai_types.Tool``. Each recognised built-in variant is mapped to the
matching ``genai_types.Tool`` field. Tools with a non-empty body after
stripping the ``type`` key (e.g. ``function_declarations``) are passed
through ``model_validate``. Variants without a ``genai_types.Tool``
equivalent (e.g. ``filesystem``, ``mcp_server``) are skipped.

Args:
agent_tools: The ``tools`` list from a fetched Gemini agent dict.

Returns:
A list of ``genai_types.Tool``, or ``None`` if there are no mappable
tools.
"""
if not agent_tools:
return None
tools: list[genai_types.Tool] = []
for tool in agent_tools:
if not isinstance(tool, dict):
continue
tool_type = tool.get("type")
if tool_type == "google_search":
tools.append(genai_types.Tool(google_search=genai_types.GoogleSearch()))
elif tool_type == "code_execution":
tools.append(
genai_types.Tool(code_execution=genai_types.ToolCodeExecution())
)
elif tool_type == "url_context":
tools.append(genai_types.Tool(url_context=genai_types.UrlContext()))
else:
# For non-built-in tools (e.g. function_declarations), strip the
# type key and validate through genai_types.Tool.
remainder = {k: v for k, v in tool.items() if k != "type"}
if remainder:
tools.append(genai_types.Tool.model_validate(remainder))
return tools or None
_agent_tools_to_config_tools = _evals_builtin_tools.agent_tools_to_config_tools


def _fetch_agent_config_dict(
Expand Down Expand Up @@ -916,7 +877,13 @@ def _fetch_agent_config_dict(
instruction = agent_dict.get("system_instruction") or None
description = agent_dict.get("description") or None
agent_type = agent_dict.get("base_agent") or None
tools = _agent_tools_to_config_tools(agent_dict.get("tools"))
has_environment = bool(
agent_dict.get("environment_config")
or agent_dict.get("base_environment")
)
tools = _agent_tools_to_config_tools(
agent_dict.get("tools"), has_environment=has_environment
)
except Exception as e: # pylint: disable=broad-exception-caught
logger.warning(
"Failed to fetch agent config for '%s' (continuing without it): %s",
Expand Down Expand Up @@ -988,33 +955,6 @@ def _agent_data_response_text(agent_data: types.evals.AgentData) -> Optional[str
return "".join(text_parts) or None


def _agent_resource_to_agent_info(
agent: str, api_client: BaseApiClient
) -> "types.evals.AgentInfo":
"""Builds an `AgentInfo` from a Gemini Agents API agent resource name.

Fetches the agent through the SDK's `api_client` (so replay recording is
preserved) via `_fetch_agent_config_dict` and derives a single-agent
`AgentInfo`: the agent's short name is the agents-map key and
`root_agent_id`.

Args:
agent: The Gemini Agents API agent resource name
(`projects/{p}/locations/{l}/agents/{name}`).
api_client: The API client used to fetch the agent.

Returns:
An `AgentInfo` describing the fetched agent.
"""
agent_config = _fetch_agent_config_dict(api_client, agent)
short_name = agent_config.agent_id
return types.evals.AgentInfo( # pytype: disable=missing-parameter
name=short_name,
agents={short_name: agent_config},
root_agent_id=short_name,
)


_INTERACTION_TERMINAL_STATES = frozenset(
["completed", "failed", "cancelled", "incomplete", "budget_exceeded"]
)
Expand Down Expand Up @@ -1108,6 +1048,11 @@ def _run_gemini_agent_inference(

interactions_client = _get_interactions_client(api_client)

# Best-effort: fetch the agent config (instruction, tools, description)
# once, so every row's agent_data carries the agents map and the display
# can render the System Topology section.
agent_config = _fetch_agent_config_dict(api_client, gemini_agent)

agent_short_id = gemini_agent.split("/")[-1]
prompts: list[str] = []
responses: list[Optional[str]] = []
Expand All @@ -1128,6 +1073,7 @@ def _run_gemini_agent_inference(
)
interaction = _await_interaction(interactions_client, interaction)
agent_data_obj = _interaction_dict_to_agent_data(interaction)
agent_data_obj.agents = {agent_config.agent_id: agent_config}
_merge_text_parts_in_agent_data(agent_data_obj)
responses.append(_agent_data_response_text(agent_data_obj))
interaction_ids.append(interaction.get("id"))
Expand Down
Loading
Loading