2025年10月29日

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


Content

こんにちは!本記事をご覧いただきありがとうございます。

企業のデータは、分析用途のデータウェアハウス(DWH)や、業務システム用のリレーショナルデータベース(RDB)など、様々な場所に散らばっていることがよくあります。これらのデータをAIエージェントからシームレスに利用できたら、業務はどれほど効率化されるでしょうか?

2025年4月にGoogleが発表した「MCP Toolbox for Databases」は、まさにそのための画期的なオープンソースツールです。
このツールは、AIがデータベースに安全かつ効率的にアクセスするための「通訳」の役割を果たします。

この記事では、MCP Toolbox for Databasesと、同じくGoogle製のAgent Development Kit (ADK)を使い、BigQueryとCloud SQL for PostgreSQLという2つの異なるデータベースから情報を横断的に検索できるAIチャットエージェントを構築する方法を、ソースコードを交えて具体的に解説します。

本記事の注目ポイントは以下の通りです。


  • YAMLファイル一つで、BigQueryとCloud SQLの両方に対応したツールを定義する手軽さ
  • 異なるデータベースの情報を、単一のAIエージェントから自然言語で問い合わせるデモ
  • 実装の容易さと、エンタープライズレベルのセキュリティ・拡張性を両立するアーキテクチャ

1. MCP Toolbox for Databasesとは

MCP Toolbox for Databases(以下、MCP Toolbox)は、AIエージェントと企業の多様なデータベース群との間に立ち、両者の複雑な連携をシンプルかつ安全に実現するための「インテリジェントな仲介役」です。これはGoogleによって開発されたオープンソースのMCPサーバーであり、AIと外部ツール(この場合はデータベース)の通信を標準化するModel Context Protocol (MCP) という規約に準拠しています。

なぜ今、MCP Toolboxが必要なのか?

AI、特に大規模言語モデル(LLM)がビジネスの中心的な役割を担うようになり、「AIに社内データベースの情報を活用させたい」というニーズが爆発的に高まっています。しかし、この一見シンプルに見える要求の裏には、開発者を悩ませる数々の技術的な障壁が存在しました。

  • 接続の複雑性: データベースごとに異なる接続ライブラリの選定、コネクションプーリングの実装、タイムアウトや再接続処理など、安定した接続を維持するだけでも多大な労力が必要でした。
  • セキュリティリスク: 認証情報の安全な管理はもちろん、最も懸念されるのが「SQLインジェクション」です。LLMが生成したSQLをそのまま実行すると、悪意のある入力によってデータベースを破壊されたり、機密情報を抜き取られたりする危険性がありました。
  • 保守性の低下: プロジェクトごとに個別の方法でデータベース連携を実装すると、システム全体が密結合になりがちです。例えば、バックエンドのデータベースをPostgreSQLからSpannerに移行するだけで、AIアプリケーション側のコードを大幅に書き直さなければならない、といった問題が発生します。

MCP Toolboxは、これらの課題を解決するために生まれました。データベース接続に関する「面倒で危険な仕事」をすべて引き受け、AIエージェント開発者が本来集中すべき「ビジネスロジックの実現」に専念できる環境を提供するのです。

「仲介役」がもたらすアーキテクチャの変革

MCP Toolboxの最大の特長は、AIエージェントとデータベースの間に抽象化レイヤーを設けるアーキテクチャにあります。

開発者は、YAML形式の設定ファイルに、どのデータベースに(sources)、どのような操作(tools)を許可するかを宣言的に記述します。するとMCP Toolboxは、その定義に基づいてAPI(ツール)を自動生成し、AIエージェントに公開します。

この「仲介役」がいることによる恩恵は計り知れません。

  • 開発時間の大幅な短縮: AIエージェント側は、接続したいデータベースがBigQueryであろうとPostgreSQLであろうと、MCP Toolboxという単一のエンドポイントと会話するだけです。データベースごとの方言や作法を学ぶ必要がなく、開発は驚くほどシンプルになります。
  • 鉄壁のセキュリティ: AIエージェントは、YAMLに定義された「ツール」しか呼び出すことができません。これは、LLMに自由なSQLを生成させるアプローチとは一線を画す「許可リスト」方式であり、SQLインジェクションのリスクを根本から排除します。認証もOAuth2やOIDCといった標準技術に対応しており、エンタープライズレベルのセキュリティを容易に実現できます。
  • 圧倒的な保守性と拡張性: 例えば、複数のデータソースから情報を取得する場合でも、tools.yamlにsource定義を追加するだけで、エージェント側のコードを変更することなく対応できます。将来的にデータベースを新しいものに移行する際も、変更箇所をtools.yamlだけに限定でき、システム全体の柔軟性が飛躍的に向上します。
  • 幅広いDB対応力: Google CloudのBigQueryやCloud SQLはもちろんのこと、オンプレミスで稼働しているPostgreSQLやMySQL、さらにはグラフデータベースのNeo4jなど、多種多様なデータベースを統一的に扱える点も大きな魅力です。

このように、MCP ToolboxはAIアプリケーション本体からデータベース接続の詳細を完全に分離します。これにより、開発の簡素化、セキュリティの強化、そして長期的な保守性の向上を同時に達成する、モダンなAIシステムアーキテクチャを実現するのです。

2. ソースコード

それでは、実際にデータ検索エージェントを構築していきましょう。今回は、BigQueryに格納された「タイタニックの乗船客室情報」と、Cloud SQL for PostgreSQLに格納された「ペンギンの個体識別情報」を検索できるエージェントを作成します。全く系統の違う2つのデータソースを検索対象とすることで、エージェントがコンテキストに応じてツールを問題なく切り替えられるかを確認しやすくするため、タイタニックとペンギンのデータを採用しました。

準備1:MCPサーバーを配置する

こちらのGitHubからMCPサーバーをダウンロードしてください。
https://github.com/googleapis/genai-toolbox

本記事では以下のフォルダ構成で構築しますので、ご参考ください。.envやライブラリに関するファイルについては割愛しております。

sample-agent/
├── agents/
│ └── tiny-agent
│ ├── agent.py
│ ├── prompt.py
│ └── sub_agents/
│ └── data_search/
│ ├── agent.py
│ ├── prompt.py
│ └── tool.py
└── mcp-toolbox/

準備2: MCPサーバー側で必要になるYAMLファイル

tools.yamlは以下です。

# tools.yaml

sources: # Defines the connection information for BigQuery # This name (bigquery_titanic_source) will be referenced by tools later bigquery_titanic_source: kind: bigquery
# IMPORTANT: Please replace with your own GCP project ID project: "YOUR_PROJECT_ID"
location: "asia-northeast1" # Tokyo region
# Defines the connection information for Cloud SQL (PostgreSQL) # Connects as localhost via the Cloud SQL Auth Proxy cloudsql_pg_source: kind: postgres
host: 127.0.0.1
port: 5433 database: "mcp-server-sample-table" user: "${DB_USER}" # Loaded from environment variable password: "${DB_PASS}" # Loaded from environment variable
tools: # Defines a tool to get the passenger list from BigQuery get_titanic_passengers: kind: bigquery-sql
source: bigquery_titanic_source # Uses the source defined above description: "Retrieves a list of passengers from the Titanic. Includes information such as survival status, age, and gender." parameters: - name: limit
type: integer
description: "Specifies the maximum number of passengers to retrieve." # The actual SQL statement to be executed # Specify the BigQuery table name statement: "SELECT * FROM `test_dataset_alpha.titanic-dataset` LIMIT @limit;"
# Defines a tool to get the penguin list from Cloud SQL get_penguins_from_cloudsql: kind: postgres-sql
source: cloudsql_pg_source # Uses the Cloud SQL source defined above description: "Retrieves penguin data from Cloud SQL (PostgreSQL)." parameters: - name: limit
type: integer
description: "Specifies the maximum number of penguins to retrieve." # The actual SQL statement to be executed statement: "SELECT * FROM penguins LIMIT $1;"
toolsets: # Groups the tools defined above # This name (titanic_analysis_tools) is referenced by the agent titanic_analysis_tools: - get_titanic_passengers
- get_penguins_from_cloudsql

各データソースに向けてsourceとtoolを設定していきます。今回はデモのため、Cloud SQLのユーザー名とパスワードは環境変数から読み取っていますが、本番環境で実装する場合はSecret Managerに格納して取得することを推奨します。

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

次に、ADKを使ってAIエージェントを実装します。MCP Toolboxで公開したツールを読み込み、エージェントに組み込みます。

ルートエージェント (agents/tiny-agent/agent.py)

# agents/tiny-agent/agent.py(ルートエージェント)

import os
import asyncio
from google.adk.agents import Agent

from .prompt import tiny_agent_instruction
from .sub_agents.data_search.agent import data_search_agent

# 環境変数の読み込み model_name = os.getenv("TINY_AGENT_MODEL")
# メインとなるルートエージェントを定義 # このエージェントが data_search_agent をサブエージェントとして持つ root_agent = Agent( model=model_name, name="tiny_agent", description="A main agent for interacting with the user.", instruction=tiny_agent_instruction, sub_agents=[ data_search_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/sub_agents/data-search/agent.py)

# agents/tiny-agent/sub_agents/data-search/agent.py(サブエージェント)
import os
from google.adk.agents import Agent
from . import tools # 自分自身のディレクトリにある tools.py をインポート from .prompts import get_data_search_instructions

# データ検索サブエージェントの定義 data_search_agent = Agent( model=os.getenv("DATA_SEARCH_AGENT_MODEL"), name="data_search_agent", description="Connects to the MCP server to retrieve structured data, such as the Titanic passenger list.", instruction=get_data_search_instructions(), tools=[ tools.get_titanic_passengers, tools.get_penguins_from_cloudsql, ], )

ツール定義 (agents/tiny-agent/sub_agents/data-search/tools.py)

# agents/tiny-agent/sub_agents/data-search/tools.py
import os
from toolbox_core import ToolboxClient
from dotenv import load_dotenv
import json

# .envファイルから環境変数を一度だけロードする load_dotenv(override=True) MCP_SERVER_URL = os.getenv("MCP_SERVER_URL")
async def get_titanic_passengers(limit: int = 5) -> str: """
タイタニック号の乗客リストを取得します。生存状況、年齢、性別などの情報が含まれます。

Args:
limit (int): 取得する乗客の最大数を指定します。デフォルトは5です。

Returns:
str: JSON形式の乗客データ、またはエラーメッセージ。
"""
print(f"サブエージェントのツール 'get_titanic_passengers' を limit={limit} で実行します。") if not MCP_SERVER_URL: return '{"error": "MCP_SERVER_URLが設定されていません。"}'
try: async with ToolboxClient(MCP_SERVER_URL) as client: # MCPサーバーで定義されている実際のツール名を指定 tool = await client.load_tool("get_titanic_passengers") results = await tool(limit=limit)
# 結果をJSON文字列に変換して返す return json.dumps(results, ensure_ascii=False, indent=2)
except Exception as e: return f'{{"error": "ツールの実行に失敗しました: {e}"}}'
async def get_penguins_from_cloudsql(limit: int = 5) -> str: """
Cloud SQL (PostgreSQL) からペンギンのデータを取得します。

Args:
limit (int): 取得するペンギンの最大数を指定します。デフォルトは5です。

Returns:
str: JSON形式のペンギンデータ、またはエラーメッセージ。
"""
print(f"サブエージェントのツール 'get_penguins_from_cloudsql' を limit={limit} で実行します。") if not MCP_SERVER_URL: return '{"error": "MCP_SERVER_URLが設定されていません。"}'
try: async with ToolboxClient(MCP_SERVER_URL) as client: # MCPサーバーで定義されている実際のツール名を指定 tool = await client.load_tool("get_penguins_from_cloudsql") results = await tool(limit=limit)
# 結果をJSON文字列に変換して返す return json.dumps(results, ensure_ascii=False, indent=2)
except Exception as e: return f'{{"error": "ツールの実行に失敗しました: {e}"}}'

MCPサーバーがデータベース接続の複雑さを吸収してくれるため、ADK側のソースコードはツールを呼び出すだけのシンプルな実装になっていることがわかります。
各エージェントへのinstruction(指示プロンプト)は別途ファイルに記載していますが、本記事では簡潔さのため割愛します。

準備4:Cloud SQLへの接続設定

Cloud SQLをローカル環境から操作するには、Cloud SQL Auth Proxyをダウンロードし、別途ターミナルを立ち上げて起動する必要があります。
お使いのOSに合わせて、こちらのGoogle Cloud公式ドキュメントからダウンロードしてください。ダウンロードした実行ファイルに決まった配置場所はありませんが、今回はプロジェクトのルート(/sample-agent)直下に配置します。

ダウンロードが完了したら、別のターミナルを立ち上げ、以下のコマンドを実行してプロキシを起動します。

.\cloud_sql_proxy.exe 'プロジェクトID:
リージョン名:インスタンス名' --port tools.yamlファイルで設定したポート番号 --credential_file='サービスアカウントキーファイルのパス'

サービスアカウントキーファイルには、「Cloud SQLクライアント」ロールを付与してください。また、サービスアカウントの作成方法の詳細は本記事では割愛しますが、Cloud SQLのインスタンスの概要から確認し、「IAMと管理のサービスアカウント」から作成してください。

準備5:MCPサーバーの起動方法

/mcp-toolbox ディレクトリ上で以下のコマンドを実行し、toolbox.exeファイルを作成してください。

$VERSION = "0.16.0"
curl.exe -L "https://storage.googleapis.com/genai-toolbox/v$VERSION/windows/amd64/toolbox.exe" -o "toolbox.exe""

tools.yamlファイルにDB_PASS、DB_USERをハードコーティングしていない場合は、以下のコマンドを実行してください。

$env:DB_USER="CLOUD_SQL_YOUR_NAME"
$env:DB_PASS="CLOUD_SQL_YOUR_PASSWORD"

最後に、以下のコマンドを実行しMCPサーバーを起動してください。

./toolbox.exe --tools-file "tools.yaml"

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

3. デモ

実際に起動させて動作確認していきましょう。`poetry run adk web`でADKのWeb UIを起動し、データ検索エージェントを起動します。
そして以下のようなリクエストをエージェントに送信すると、エージェントが検索したいデータに合わせて参照画像のように
データを取得し、レスポンスを返してくれます。
タイタニックのデータはBigQueryに格納されていることをプロンプトで予め記載しているので問題なく、BigQueryのツールを使って取得できています。
ペンギンのデータも同様にCloud SQLから問題なく取得できています。
以上のように、無事実装が成功していることを確認できましたね!

4. まとめ

本記事では、MCP Toolbox for Databasesと**Agent Development Kit (ADK)**を用いて、BigQueryとCloud SQL for PostgreSQLという2つの異なるデータベースを横断的に検索するAIエージェントを構築しました。

今回の実装から、以下のメリットが確認できました。

  • 驚異的な実装の容易さ: データベースごとの接続ロジックを実装することなく、tools.yamlという単一の設定ファイルで、複数のデータソースを「ツール」として迅速に定義できました。
  • 統一的なインターフェース: エージェント側は、データの取得元がBigQueryかCloud SQLかを意識する必要がありません。MCP Toolboxがその差異を吸収し、統一的なツールとして提供してくれるため、エージェントの実装が非常にシンプルになります。
  • 高い制御性と安全性: YAMLで許可した操作(ツール)しか実行できないため、AIエージェントの自由な振る舞いを制限し、意図しないSQLの実行を防ぐことができます。これはエンタープライズ環境において極めて重要です。

一方で、参照記事でも触れられているように、この「ツールによる制御」は、アドホックな分析のような自由なデータ操作とはトレードオフの関係にあります。しかし、多くの業務アプリケーションでは、定型的なデータ検索・操作が中心となるため、MCP Toolboxのアプローチは非常に有効です。

MCP Toolbox for Databasesは、企業内に散在する貴重なデータを、AIエージェントを通じて安全かつ手軽に活用するための強力なソリューションです。ぜひ、皆さんの環境でもお試しください。

2025年10月29日 BigQueryとCloud SQLを横断検索!MCP ToolboxとADKで作るハイブリッドデータ検索エージェント

Category Google Cloud

ご意見・ご相談・料金のお見積もりなど、
お気軽にお問い合わせください。

お問い合わせはこちら