Gemini Enterprise でデータ可視化マルチエージェントを構築： CData Connect AI MCP × ADK BuiltInCodeExecutor

by 杉本和也 | April 21, 2026 | Last Updated: April 21, 2026

こんにちは。CData Software Japan リードエンジニアの杉本です。

以前までの記事でGemini Enterprise でCData Connect AI を使ったエージェントのシンプルな開発方法を解説してきました。

Gemini Enterprise から CData Connect AI に OAuth で接続する方法：カスタムOAuth アプリを活用した認証・認可の設定手順
https://jp.cdata.com/blog/gemini-enterprise-connect-ai-oauth

ただ、現状Gemini Enterprise はMarkdown か画像出力のみがアウトプットの範囲のようで、Claude でHTML・JavaScript を組み合わせたインタラクティブなダッシュボードを作成するようなことは難しい感じでした・・・。

とはいえ、MCP 経由で取得した結果から適切なインサイトを得るうえで、グラフなどのビジュアライズの表現は重要ですよね。実はそんな時にGemini Enterprise・Vertex AI ではPython のプログラムを実行してグラフの生成などを行うことができる「BuiltInCodeExecutor」という機能があります！

そこで、今回は Google の Agent Development Kit（ADK）を使って、CData Connect AI MCP でデータを取得し、Gemini の BuiltInCodeExecutor でグラフ描画まで一気通貫で行うマルチエージェントをGemini Enterprise・Vertex AI 上構築してみました！

これを利用することで、以下のようにチャット上にアドホックにグラフなどを生成して確認することができるようになります。

また、MCP を経由してデータのやり取りをしている際、Gemini Enterprise の画面上はずっとローディング画面になって反応があまり無く、うまく動いているのか心配になってしまうことがありました。

そこでMCP のツールコールのやり取りも途中経過のプロセスとして確認できる実装も施しています。

実装自体はシンプルに見えるのですが、実際に動かすといくつかのハマりポイントがありました。この記事では構築手順だけでなく、そのトラブルシューティングの過程も詳しく共有したいと思います。Google ADK でマルチエージェントを開発している方にとって参考になれば幸いです。

対象読者

Python と Google ADK の基本的な使い方を知っている開発者を対象としていますが、サンプルコードも記載しているので、そのままデプロイしてもらっても手軽に試していただけます。

CData Connect AI の基礎知識があると理解しやすいですが、トラブルシューティングの内容だけでも ADK 開発者に参考になる部分があるかと思います。

必要なもの

今回使用するサービス・ライブラリは以下のとおりです。

CData Connect AI（MCP エンドポイントを提供）
Google ADK（google-adk パッケージ）
Gemini 2.5 Pro（gemini-2.5-pro）
Gemini Enterprise（トライアルあり：https://jp.cdata.com/blog/gemini-enterprise-trial）

それぞれトライアル環境を取得して構築しました。

全体アーキテクチャ

今回は以下の3層構成でマルチエージェントを実装しました。

RootAgent（BaseAgent 継承）: ツールコール・レスポンスのリアルタイム表示を担当するオーケストレータ
connect_ai_code_execution_inner（Agent）: CData Connect AI MCP ツールと viz_agent を呼び分けるオーケストレータ LLM
viz_agent（Agent + BuiltInCodeExecutor）: matplotlib で Python コードを実行してグラフを生成するサブエージェント

この3層構成を選んだ背景には、Google ADK のツール制約があります。BuiltInCodeExecutor（組み込みコード実行）は、同じエージェント内で他のツールと共存できません。つまり、code_executor=BuiltInCodeExecutor() を設定したエージェントに McpToolset などのツールを同時に追加することができないのです。

# NG: BuiltInCodeExecutor と MCP ツールは同一エージェントに共存できない
agent = Agent(
    tools=[McpToolset(...)],          # MCP ツール
    code_executor=BuiltInCodeExecutor()  # <-- サポート外
)

この制約は ADK の公式ドキュメント Tool limitations にも明記されており、公式の回避策として「別々のエージェントに分離して AgentTool で組み合わせる」マルチエージェント構成が推奨されています。今回の実装はまさにこのパターンを採用しています。

実装手順

今回実装したソースコードの全体は以下のとおりです。
なお、CData Connect AI への認証方式にはわかりやすいようにBasic 認証を用いました。
Basic 認証用の接続情報は.env で管理しています。

Vertex AI・Gemini Enterpriseへの展開方法は以下の記事を参考にしてみてください。

Gemini Enterprise でCData Connect AI のリモートMCP と連携可能なエージェントを開発する方法
https://jp.cdata.com/blog/gemini-enterprise-cdata-connect-ai
``


import base64
import json
import os
from typing import AsyncGenerator

from google.adk.agents import Agent, BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.agents.readonly_context import ReadonlyContext
from google.adk.code_executors import BuiltInCodeExecutor
from google.adk.events import Event
from google.adk.events.event_actions import EventActions
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset, StreamableHTTPConnectionParams
from google.genai import types

MCP_URL = os.getenv("CDATA_CONNECT_AI_MCP_URL", "https://mcp.cloud.cdata.com/mcp")
_BASIC_AUTH_USERNAME = os.getenv("CDATA_CONNECT_AI_USERNAME", "")
_BASIC_AUTH_PASSWORD = os.getenv("CDATA_CONNECT_AI_PASSWORD", "")
MODEL = "gemini-2.5-pro"

def _get_basic_auth_headers(context: ReadonlyContext) -> dict[str, str]:
    credentials = base64.b64encode(
        f"{_BASIC_AUTH_USERNAME}:{_BASIC_AUTH_PASSWORD}".encode()
    ).decode()
    return {"Authorization": f"Basic {credentials}"}


_viz_agent = Agent(
    name="viz_agent",
    model=MODEL,
    description="Pythonコードを実行してデータを可視化するエージェント",
    instruction=(
        "あなたはデータを可視化するエージェントです。"
        "渡されたデータを matplotlib を使った Python コードで可視化してください。"
    ),
    code_executor=BuiltInCodeExecutor(),
)

_root_llm_agent = Agent(
    name="connect_ai_code_execution_inner",
    model=MODEL,
    description="CData Connect AI + Code Execution オーケストレータ（内部）",
    instruction=(
        "あなたは CData Connect AI とデータ可視化を組み合わせたアシスタントです。\n"
        "ユーザーのリクエストに応じて適切に処理してください:\n"
        "- データの取得・確認が必要な場合: CData Connect AI MCP ツールを呼び出す\n"
        "- データの可視化が必要な場合: まず MCP ツールでデータを取得し、"
        "その結果を viz_agent に渡して可視化する\n"
        "- 取得済みデータを可視化する場合: viz_agent だけ呼び出す\n\n"
        "【viz_agent への渡し方】\n"
        "viz_agent を呼び出す際は、JSON のまま渡さず、必ずデータを CSV 形式のテキストに変換してから渡すこと。\n"
        "request パラメータには、以下の形式で文字列として記述すること:\n"
        "- 1行目: 可視化の指示（例: '会社別売上の棒グラフを作成してください'）\n"
        "- 2行目以降: CSV 形式のデータ（ヘッダー行 + データ行）\n"
        "例:\n"
        "会社名ごとの売上棒グラフを作成してください。タイトルは「顧客別売上合計」。\n"
        "会社名,売上\n"
        "A社,100\n"
        "B社,200\n"
    ),
    tools=[
        McpToolset(
            connection_params=StreamableHTTPConnectionParams(url=MCP_URL),
            header_provider=_get_basic_auth_headers,
        ),
        AgentTool(agent=_viz_agent),
    ],
)


class RootAgent(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        async for event in _root_llm_agent.run_async(ctx):
            func_calls = event.get_function_calls()
            func_resps = event.get_function_responses()

            if func_calls:
                yield event
                for func_call in func_calls:
                    args_str = json.dumps(func_call.args, ensure_ascii=False, indent=2)
                    yield Event(
                        invocation_id=ctx.invocation_id,
                        author=self.name,
                        partial=True,  # session.events に保存されず "For context:" 注入を防ぐ
                        content=types.Content(
                            role="model",
                            parts=[types.Part(text=f"\n**[Tool Call >]** `{func_call.name}`\n```json\n{args_str}\n```\n**[< Tool Call]**\n\n")],
                        ),
                    )
            elif func_resps:
                yield event
                for func_resp in func_resps:
                    # viz_agent の結果は画像アーティファクトで表示するため JSON 表示をスキップ
                    if func_resp.name != _viz_agent.name:
                        result_str = json.dumps(func_resp.response, ensure_ascii=False, indent=2)
                        yield Event(
                            invocation_id=ctx.invocation_id,
                            author=self.name,
                            partial=True,
                            content=types.Content(
                                role="model",
                                parts=[types.Part(text=f"\n**[Tool Result >]** `{func_resp.name}`\n```json\n{result_str}\n```\n**[< Tool Result]**\n\n")],
                            ),
                        )
            else:
                yield event


root_agent = RootAgent(
    name="connect_ai_code_execution",
    description="CData Connect AI + Code Execution オーケストレータ",
    sub_agents=[_root_llm_agent, _viz_agent],
)

.env

CDATA_CONNECT_AI_MCP_URL=https://mcp.cloud.cdata.com/mcp
CDATA_CONNECT_AI_USERNAME=XXXX
CDATA_CONNECT_AI_PASSWORD=XXXX

init.py

from . import agent

viz_agent の定義

それではここから各コードの詳細を簡単に解説していきます。

まず、Gemini の組み込みコード実行機能（BuiltInCodeExecutor）を使ってグラフを描画するエージェントを定義します。

from google.adk.agents import Agent
from google.adk.code_executors import BuiltInCodeExecutor

_viz_agent = Agent(
    name="viz_agent",
    model="gemini-2.5-pro",
    description="Pythonコードを実行してデータを可視化するエージェント",
    instruction=(
        "あなたはデータを可視化するエージェントです。"
        "渡されたデータを matplotlib を使った Python コードで可視化してください。"
    ),
    code_executor=BuiltInCodeExecutor(),
)

BuiltInCodeExecutor は Gemini 2.0 以降で使用できる機能で、Gemini が生成した Python コードをサンドボックス環境で実行し、結果（グラフ画像など）をアーティファクトとして返してくれます。

https://adk.dev/integrations/code-execution/

メイン LLM エージェントの定義

次に、CData Connect AI MCP ツールと viz_agent を組み合わせてオーケストレートするエージェントを定義します。tools にCData Connect AI のMCP と viz_agent を利用するためのAgentTool をパラメータとして設定しています。

import base64
import os
from google.adk.agents import Agent
from google.adk.agents.readonly_context import ReadonlyContext
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset, StreamableHTTPConnectionParams

MCP_URL = os.getenv("CDATA_CONNECT_AI_MCP_URL", "https://mcp.cloud.cdata.com/mcp")
_BASIC_AUTH_USERNAME = os.getenv("CDATA_CONNECT_AI_USERNAME", "")
_BASIC_AUTH_PASSWORD = os.getenv("CDATA_CONNECT_AI_PASSWORD", "")

def _get_basic_auth_headers(context: ReadonlyContext) -> dict[str, str]:
    credentials = base64.b64encode(
        f"{_BASIC_AUTH_USERNAME}:{_BASIC_AUTH_PASSWORD}".encode()
    ).decode()
    return {"Authorization": f"Basic {credentials}"}

_root_llm_agent = Agent(
    name="connect_ai_code_execution_inner",
    model="gemini-2.5-pro",
    description="CData Connect AI + Code Execution オーケストレータ（内部）",
    instruction=(
        "あなたは CData Connect AI とデータ可視化を組み合わせたアシスタントです。\n"
        "ユーザーのリクエストに応じて適切に処理してください:\n"
        "- データの取得・確認が必要な場合: CData Connect AI MCP ツールを呼び出す\n"
        "- データの可視化が必要な場合: まず MCP ツールでデータを取得し、"
        "その結果を viz_agent に渡して可視化する\n\n"
        "【viz_agent への渡し方】\n"
        "viz_agent を呼び出す際は、JSON のまま渡さず、必ずデータを CSV 形式のテキストに変換してから渡すこと。\n"
        "request パラメータには、以下の形式で文字列として記述すること:\n"
        "- 1行目: 可視化の指示（例: '会社別売上の棒グラフを作成してください'）\n"
        "- 2行目以降: CSV 形式のデータ（ヘッダー行 + データ行）\n"
    ),
    tools=[
        McpToolset(
            connection_params=StreamableHTTPConnectionParams(url=MCP_URL),
            header_provider=_get_basic_auth_headers,
        ),
        AgentTool(agent=_viz_agent),
    ],
)

McpToolset に header_provider を指定することで、リクエストごとに動的に認証ヘッダーを生成できます。CData Connect AI MCP への接続には Basic 認証を使用しています。

RootAgent の実装

最後に、ツールコール・レスポンスをリアルタイム表示するための RootAgent を実装します。


class RootAgent(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        async for event in _root_llm_agent.run_async(ctx):
            func_calls = event.get_function_calls()
            func_resps = event.get_function_responses()

            if func_calls:
                yield event
                for func_call in func_calls:
                    args_str = json.dumps(func_call.args, ensure_ascii=False, indent=2)
                    yield Event(
                        invocation_id=ctx.invocation_id,
                        author=self.name,
                        partial=True,  # session.events に保存されず "For context:" 注入を防ぐ
                        content=types.Content(
                            role="model",
                            parts=[types.Part(text=f"\n**[Tool Call >]** `{func_call.name}`\n```json\n{args_str}\n```\n**[< Tool Call]**\n\n")],
                        ),
                    )
            elif func_resps:
                yield event
                for func_resp in func_resps:
                    # viz_agent の結果は画像アーティファクトで表示するため JSON 表示をスキップ
                    if func_resp.name != _viz_agent.name:
                        result_str = json.dumps(func_resp.response, ensure_ascii=False, indent=2)
                        yield Event(
                            invocation_id=ctx.invocation_id,
                            author=self.name,
                            partial=True,
                            content=types.Content(
                                role="model",
                                parts=[types.Part(text=f"\n**[Tool Result >]** `{func_resp.name}`\n```json\n{result_str}\n```\n**[< Tool Result]**\n\n")],
                            ),
                        )
            else:
                yield event

ここでの実装にはいくつかの工夫が含まれています。次のセクションで詳しく説明します。

ハマったポイントと解決策

実装中に2つの問題に直面しました。それぞれの原因と解決策を共有します。

1. カスタムイベントが内部エージェントに注入されてエージェントが止まる

現象

ツールコール結果をリアルタイム表示するために、RootAgent からカスタムイベントを yield したところ、途中でエージェントが応答を返さなくなりました。エラーメッセージは出ておらず、ログを見ると finish_reason: STOP で空のテキストが返ってきた状態でした。

原因

ADK のソースコード（contents.py の _present_other_agent_message()）を調査したところ、session.events に保存されている他エージェントのイベントが「For context:」というプレフィックスを付けて内部エージェントへの user メッセージとして注入される仕組みになっていました。

RootAgent が yield したカスタムイベントがそのまま session.events に保存され、次のターンで _root_llm_agent のプロンプトに自分自身のカスタムメッセージが注入される → モデルが混乱して空テキストを返す、というループに陥っていたのです。

ADK のイベントの仕組みについては Events - Agent Development Kit および Event Loop - Agent Development Kit に詳しく記載されています。

解決策

カスタムイベントに partial=True を設定することで解決しました。

yield Event(
    invocation_id=ctx.invocation_id,
    author=self.name,
    partial=True,  # ← これが重要
    content=types.Content(...),
)

ADK の append_event は partial=True のイベントを受け取ると早期 return し、session.events への保存をスキップします。これにより、カスタムイベントが次のターンのプロンプトに注入されなくなります。

2. MALFORMED_FUNCTION_CALL でエージェントが停止する

現象

CData Connect AI MCP でデータを取得した後、viz_agent への受け渡し時に MALFORMED_FUNCTION_CALL エラーが発生してエージェントが止まることがありました。

原因

queryData ツールで取得した大量の JSON データ（数百行のレコード）をそのまま viz_agent の request 引数として渡そうとしたとき、モデルが巨大な JSON をシリアライズした関数呼び出しを生成しようとして不正な関数呼び出しになっていました。

解決策

_root_llm_agent の instruction に「viz_agent への渡し方」のルールを追加し、データを CSV 形式のテキストに変換してから渡すよう指示しました。

【viz_agent への渡し方】
viz_agent を呼び出す際は、JSON のまま渡さず、必ずデータを CSV 形式のテキストに変換してから渡すこと。
request パラメータには、以下の形式で文字列として記述すること:
- 1行目: 可視化の指示（例: '会社別売上の棒グラフを作成してください'）
- 2行目以降: CSV 形式のデータ（ヘッダー行 + データ行）

JSON から CSV への変換はモデルが行うため、プログラム側に変換処理を追加する必要はありません。これにより MALFORMED_FUNCTION_CALL が解消されました。

動作確認

実際に「CData Connect AI で取得した売上データをグラフにして」と入力すると、以下の流れで処理が進みます。

connect_ai_code_execution_inner が CData Connect AI MCP の queryData ツールを呼び出してデータを取得
取得したデータを CSV 形式に変換して viz_agent に渡す
viz_agent が matplotlib コードを生成・実行してグラフ画像を生成
RootAgent がグラフ画像のファイル名を検出して artifact_delta を emit
UI にグラフ画像が表示される

以下のようにツールコールとその結果、最終的なグラフ画像がリアルタイムで表示されていることが確認できました！

おわりに

このように Google ADK の BaseAgent と Gemini の BuiltInCodeExecutor、CData Connect AI MCP を組み合わせることで、自然言語でデータの取得からグラフ描画まで一気通貫で行うエージェントを構築できます。

ADK・Vertex AI で構築したエージェントをGemini Enterprise で展開する場合、フロントエンドで色々と制約が発生するので、今回のような工夫が色々と必要になってきますね。

ソリューションとユースケース CData Connect AI CData Connect AI MCP ADK Gemini Enterprise Vertex AI

CData は、AI を本番環境で機能させるデータレイヤーです。350 以上のソースへのライブコネクティビティとレプリケーション、セマンティックコンテキスト、組み込みガバナンスを提供。Databricks、Microsoft、Google、Palantir をはじめ、世界中の 10,000 以上のお客様の AI を支えています。

お問い合わせ

Blog