今回使用するBuiltInCodeExecutorは、エージェントにPythonコードの実行能力を与えるための組み込みツールです。

具体的には、以下の特徴を持っています。

• LLMによるコード生成: エージェント（内部的にはGeminiモデル）が、ユーザーからの指示を解釈して、タスクを解決するためのPythonコードを自動で生成します。
• 安全な実行環境: 生成されたコードは、安全なサンドボックス環境で実行されます。これにより、計算やデータ操作などをセキュアに行うことが可能です。
• 正確なタスク処理: LLM単体では間違いやすい数値計算や、順序立てたデータ処理などを、Pythonに任せることで正確に実行できます。

例えば、ユーザーが「(5 + 7) * 3は？」と質問したとします。このとき、エージェントは単に答えを推測するのではなく、print((5 + 7) * 3)というPythonコードを内部で生成・実行し、その出力である「３６」を確実な答えとしてユーザーに返すのです。

このように、BuiltInCodeExecutorを組み込むだけで、エージェントの能力を飛躍的に向上させることができます。
今回紹介する組み込みツールのユースケースとしては、「データサイエンスエージェントに組み込んでPythonを使った高度な分析」を行うことです。
Pythonにはscikit-learnライブラリがあるのでエージェントに、このライブラリを使用させ、LLMで行うよりも質の高い分析を可能にすることができるでしょう。
更に、分析に使用したプログラムコードを発行できるため、LLMが行った分析と違い分析の中身を確認することができたり、チーム間で共有し合えることも大きなメリットであると感じております。

それでは、実際にPythonを実行するデータ分析エージェントを構築していきましょう。今回は、BigQueryに格納された「タイタニックの乗船客室情報」を取得し、それを基にデータ分析を行います。

組み込みツール自体はサブエージェントではなく、データ分析エージェンのAgentToolとして実装します。理由としては、組み込みツールのこちらの制限事項を回避するためです。

• 現在、各ルートエージェントまたは単一のエージェントに対して、サポートされている組み込みツールは1つだけです。同じエージェント内で他のどのタイプのツールも使用することはできません。
• 組み込みツールはサブエージェント内では使用できません。

参考ドキュメント：組み込みツール – Agent Development Kit

単純に組み込みツールとして使うと制限事項に引っかかるのを回避したいのです。
ちなみに自身で基となるライブラリを読み込んでツールを作成するやり方がありますが、単なるツールではなく、AgentToolとして実装することでPythonコードの生成の精度が上がるのではないかという狙いがあります。
また、BigQueryからのデータ取得の詳しい方法はこちらの記事をご参照ください。

BigQueryとCloud SQLを横断検索！MCP ToolboxとADKで作るハイブリッドデータ検索エージェント

準備1：フォルダ構成

以下のような構成になっております。.envファイルや.venvファイル、データ取得のためのMCPサーバー関連は本記事の内容から逸れてしまうので、割愛します。

sample-agent/

└── agents/

    └── tiny-agent

        ├── agent.py

        ├── prompt.py

        └── sub_agents/

             ├── data_analysis/

             │    ├── agent.py

             │    ├── prompt.py

             │    └── tool.py

             └── code_interpreter/

                  ├── agent.py

                  ├── prompt.py

                  └── tool.py

準備3：ADKでエージェントを実装する

次に、ADKを使ってAIエージェントを実装します。

agents/tiny-agent/agent.py（ルートエージェント）

import os

from google.adk.agents import Agent

from .prompt import tiny_agent_instruction

from .sub_agents.data_analysis.agent import data_analysis_agent



# 環境変数の読み込み
model_name = os.getenv("TINY_AGENT_MODEL")


# メインとなるルートエージェントを定義
# データ分析の全ての処理は data_analysis_agent に委任する
root_agent = Agent(
    model=model_name,
    name="tiny_agent",
    description="A main agent that delegates data analysis tasks to a specialist agent.",
    instruction=tiny_agent_instruction,
    sub_agents=[
        data_analysis_agent,
    ],
)




# 対話形式でエージェントをテストするための実行ブロック
async def main():
    print("エージェントを起動します。終了するには 'exit' と入力してください。")
    while True:
        user_input = input("You: ")
        if user_input.lower() == 'exit':
            break
        

        response_gen = root_agent.generate(prompt=user_input)
        

        full_response = ""
        async for chunk in response_gen:
            if "content" in chunk:
                print(chunk["content"], end="", flush=True)
                full_response += chunk["content"]
            elif "tool_code" in chunk:
                print(f"\n--- Executing Tool: {chunk['tool_code']}\n")


if __name__ == "__main__":
    asyncio.run(main())

agents/tiny-agent/prompt.py

# Copyright 2025 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.


tiny_agent_instruction = '''You are a helpful and friendly AI assistant.

Your primary role is to coordinate tasks between specialized sub-agents and to communicate with the user.



**Your Sub-Agents:**

- **`data_search_agent`**: Use this agent when the user asks for specific data, such as "Titanic passenger list" or "penguin data."

- **`code_interpreter_agent`**: Use this agent when the user wants to analyze data that has already been retrieved. This agent can execute Python code to perform calculations, analysis, or transformations on the data.



**Workflow:**

1.  **General Conversation**: For casual chat or questions that don't involve data, respond directly to the user in a friendly manner.

2.  **Data Retrieval**: If the user's request is clearly about retrieving data (e.g., "get," "find," "search for data"), you **must** delegate this task to the `data_search_agent`.

3.  **Data Analysis**: If the user asks to analyze, compute, or transform the data that was just retrieved (e.g., "calculate the average age," "group by survival status"), you **must** delegate this task to the `code_interpreter_agent`. The `code_interpreter_agent` will automatically use the data from the last query.



**Important Rules:**

- **Stick to your role.** Do not perform data searches or code execution yourself. Always delegate to the appropriate sub-agent.

- **One task at a time.** Handle either a data search or a data analysis request in a single turn.

- **Inform the user.** Let the user know which agent you are using and what you are doing. For example, say "I will ask the data search agent to retrieve the Titanic passenger list." or "I will ask the code interpreter to calculate the average age."

'''

agents/tiny-agent/sub_agents/data_analysis/agent.py

import os

from google.adk.agents import Agent

from google.adk.tools import AgentTool

from . import tools

from . import prompts

from ..code_interpreter.agent import code_interpreter_agent



data_analysis_agent = Agent(
    model=os.getenv("DATA_ANALYSIS_AGENT_MODEL"),
    name="data_analysis_agent",
    instruction=prompts.get_data_analysis_instructions(),
    tools=[
        tools.get_titanic_passengers,
        tools.get_penguins_from_cloudsql,
        tools.preview_data_from_state,
        AgentTool(agent=code_interpreter_agent),
    ],
)

agents/tiny-agent/sub_agents/data_analysis/prompts.py

def get_data_analysis_instructions() -> str:
    """Returns the instruction string for the DataAnalysisAgent."""
    return """

    You are a self-sufficient data analyst. Your goal is to answer user questions by fetching and analyzing data.



    **Your Tools:**

    1.  **Data Retrieval Tools** (`get_titanic_passengers`, `get_penguins_from_cloudsql`): Use these to fetch data. The results are saved to your internal state under the key 'last_query_result'.

    2.  `preview_data_from_state`: After fetching data, use this to inspect its structure.

    3.  `code_interpreter_agent`: Your specialist tool for executing Python code.



    **Workflow:**

    1.  Use a data retrieval tool to get the data.

    2.  After the tool runs, the data is available to you in the state. You can see a preview of it below.

    3.  Based on the data preview, write the Python code needed for the analysis.

    4.  **CRITICAL:** Call the `code_interpreter_agent` tool. You MUST pass the data from your state to its `json_data_string` argument. Use the placeholder like this: `code_interpreter_agent(python_code="...", json_data_string="{last_query_result?}")`.

        **IMPORTANT**: The tool call must be a direct call to `code_interpreter_agent`. Do not wrap it in `print()` or any other function.

    5.  Interpret the results and provide a final answer.



    **Data Preview from last query:**

    {last_query_result?}

    """

agents/tiny-agent/sub_agents/data_analysis/tools.py

import os

import json

import pandas as pd

import io

from google.adk.tools import ToolContext

from toolbox_core import ToolboxClient

from dotenv import load_dotenv



# Load environment variables for MCP_SERVER_URL
load_dotenv(override=True)
MCP_SERVER_URL = os.getenv("MCP_SERVER_URL")


__all__ = [
    "get_titanic_passengers",
    "get_penguins_from_cloudsql",
    "preview_data_from_state",
]


# --- Data Retrieval Tools ---


async def get_titanic_passengers(limit: int = 100,  tool_context: "ToolContext | None" = None,) -> str:
    """

    Retrieves the Titanic passenger list and saves the result to the agent's internal state.

    """
    if not MCP_SERVER_URL: return '{"error": "MCP_SERVER_URL is not set."}'
    if not tool_context: return '{"error": "Tool context is not available."}'
    try:
        async with ToolboxClient(MCP_SERVER_URL) as client:
            tool = await client.load_tool("get_titanic_passengers")
            results = await tool(limit=limit)
            json_results = json.dumps(results, ensure_ascii=False)
            tool_context.state["last_query_result"] = json_results

            return f"Successfully retrieved {len(results)} passenger records and saved them to state."
    except Exception as e:
        return f'{{"error": "Failed to execute tool: {e}""}}'


async def get_penguins_from_cloudsql(limit: int = 100, tool_context: "ToolContext | None" = None,) -> str:
    """

    Retrieves penguin data from Cloud SQL and saves the result to the agent's internal state.

    """
    if not MCP_SERVER_URL: return '{"error": "MCP_SERVER_URL is not set."}'
    if not tool_context: return '{"error": "Tool context is not available."}'
    try:
        async with ToolboxClient(MCP_SERVER_URL) as client:
            tool = await client.load_tool("get_penguins_from_cloudsql")
            results = await tool(limit=limit)
            json_results = json.dumps(results, ensure_ascii=False)
            tool_context.state["last_query_result"] = json_results

            return f"Successfully retrieved {len(results)} penguin records and saved them to state."
    except Exception as e:
        return f'{{"error": "Failed to execute tool: {e}""}}'


# --- State Preview Tool ---


def preview_data_from_state(tool_context: "ToolContext | None" = None) -> str:
    """

    Previews the data saved in 'last_query_result' from the agent's state.

    """
    if not tool_context or "last_query_result" not in tool_context.state:
        return "Error: Data not found in state. Please run a query first."
    try:
        df = pd.read_json(io.StringIO(tool_context.state["last_query_result"]))
        if df.empty: return "The query returned no data."
        return f"Preview of data:\n\n**Columns:**\n{df.columns.tolist()}\n\n**First 3 rows:**\n{df.head(3).to_string()}"
    except Exception as e:
        return f"Error processing data from state: {e}"

agents/tiny-agent/sub_agents/code_interpreter/agent.py(コード生成・実行サブエージェント)

import os

from google.adk.agents import Agent

from . import tools

from . import prompts



code_interpreter_agent = Agent(
    model=os.getenv("CODE_INTERPRETER_AGENT_MODEL"),
    name="code_interpreter_agent",
    instruction=prompts.get_code_interpreter_instructions(),
    description="""

    A specialized agent that executes Python code.

    It takes a JSON string of data and a Python script as input,

    runs the script with the data loaded into a pandas DataFrame named 'df',

    and returns the execution result (stdout or stderr).

    This agent is used as a tool by other agents.

    """,
    tools=[
        tools.execute_python_code,
    ],
)

agents/tiny-agent/sub_agents/code_interpreter/prompts.py

def get_code_interpreter_instructions() -> str:
    """Returns the instruction string for the CodeInterpreterAgent."""
    return """

    You are a specialized agent that executes Python code.

    Your purpose is to run the `execute_python_code` tool with the provided arguments.

    Do not engage in conversation.

    """

agents/tiny-agent/sub_agents/code_interpreter/tools.py

import logging

from google.adk.tools import ToolContext

from google.adk.code_executors import BuiltInCodeExecutor

from google.adk.code_executors.code_execution_utils import CodeExecutionInput



def execute_python_code(
    python_code: str,
    json_data_string: str = "",
    tool_context: "ToolContext | None" = None,
) -> str:
    """

    Injects JSON data into a Python script as a pandas DataFrame and executes it

    to return the analysis result.

    The data can be passed directly as a JSON string or loaded from the agent's state.



    Args:

        python_code: The Python code to execute for analysis.

                     This code should assume a pandas DataFrame named `df` is available.

        json_data_string: Optional. A JSON string representing the data to be analyzed.

                          If not provided (or empty), the tool will attempt to load data

                          from the 'last_query_result' key in the agent's state.

        tool_context: The context object for the current tool call.



    Returns:

        A string containing the standard output and/or standard error from the

        Python script execution.

    """
    if not tool_context:
        return "Error: Tool context is not available. Cannot manage session state."


    # If data is not passed directly, try to load it from state
    if not json_data_string:
        if "last_query_result" in tool_context.state:
            json_data_string = tool_context.state["last_query_result"]
        else:
            return "Error: No data provided. Please run a query first or pass the data directly."


    # Prepare the full Python script by injecting the data loading code
    data_loading_code = f"""

import pandas as pd

import io

import json



# Data is injected here

json_data = '''

{json_data_string}
'''

df = pd.read_json(io.StringIO(json_data))

"""
    full_python_code = data_loading_code + "\n" + python_code



    # Use BuiltInCodeExecutor for local execution
    executor = BuiltInCodeExecutor()
    try:
        result = executor.execute_code(
            invocation_context=tool_context,
            code_execution_input=CodeExecutionInput(code=full_python_code),
        )
        if result is None:
            return "Code executed successfully with no output."
        if result.stderr:
            logging.error(f"Python execution error: {result.stderr}")
            return f"Python Execution Error:\n{result.stderr}"
        return result.stdout



    except Exception as e:
        logging.error(f"Error during code execution: {e}", exc_info=True)
        return f"Error: An unexpected error occurred during Python code execution. Details: {e}"

こちらで準備完了です。後はエージェントを起動させればデモプレイが可能です。
今回私は、poetryコマンドを用いてデモプレイを行いました。

実際に起動させて動作確認してきましょう。poetry run adk webでADKのWeb UIを起動し、データ検索エージェントを起動します。
そして以下のリクエストを送信すると、エージェントがデータ取得、Pythonを使ったデータ分析を行ってくれます。

問題なくデータ分析してくれます。また上記にもある通り、分析に使用したプログラムコードもエージェントにリクエストすれば
確認することができます。

このように、Pythonをサンドボックス上で実行しているため、分析過程を可視化できます。

ブラックボックス化しないのは非常にありがたいです。根拠のある分析ができる他、実際に上記のプログラムコードを自身で実行、カスタマイズできるのも

魅力の１つです。

本記事では、ADKの組み込みツールである BuiltInCodeExecutorを使用し、Pythonで高度なデータ分析を行うエージェントを作成しました。

今回のポイントをまとめます。

• ADKの BuiltInCodeExecutor を使うと、エージェントにPythonコードの実行能力を簡単に付与できる。
• LLMだけでは困難な高度なデータ分析や機械学習モデルの構築も、scikit-learnなどのライブラリを活用したPythonコードを生成・実行させることで解決できる。
• 生成されたPythonコードを確認できるため、分析プロセスの透明性が高く、結果の再現性も確保できる。これはチームでの分析結果の共有にも非常に役立つ。
• 他のツールと組み合わせることで、データ取得から分析、報告までの一連のワークフローを自動化する、より実践的なエージェントを構築可能になる。

今回は線形回帰モデルを試しましたが、この仕組みを応用すれば、クラスタリングによる顧客セグメンテーションや、分類モデルによる予測タスクなど、さらに多様なデータサイエンスの課題に取り組むことができます。