MCPとAIエージェントを活用してSlackから顧客情報を横断的に検索できるようにした話

こんにちは。データエンジニアの田頭(tagasyksk)です。本記事では、MCPとAIエージェントを活用して、複数CRMの顧客情報を横断的に検索できるようにした事例をご紹介します。

背景
システム構成図
技術選定
工夫した点
導入後の効果
今後の展望
終わりに

背景

ファインディでは、エンジニア組織をあらゆる場面で支援するため、複数のプロダクトを展開しています。

事業成長に伴う課題として、お客様の大切な情報がプロダクト毎にサイロ化してしまう状況が起きました。
そこで、この課題を解消し、一社でも多くのお客様にファインディの価値を届けるため、CRMに蓄積されていた顧客情報をBigQueryに集約し、部門横断で利用できる「共通企業マスタ」を構築しました。これにより、全部門が常に同じ最新の顧客情報を参照できるようになり、スムーズな連携が可能な体制を整えました。

しかし、この共通企業マスタをどのようにセールスチームやCSチームに届けるかが新たな課題となりました。全員にSQLを書かせるのは現実的でなく、新しい分析インターフェースを作るのも工数がかかります。そこで、MCPとAIエージェントを活用したSlack上での企業検索システムの構築に取り組みました。

システム構成図

Slackとソケットモードを利用して疎通するAIエージェントサーバーと、後述するBigQueryのMCPサーバーをCloud Runのマルチコンテナ構成を利用してホスティングしています。

技術選定

AIエージェントのフレームワークには、GoogleからリリースされているAgent Development Kit(ADK)を採用しました。

google.github.io

ADKを選定した理由は次の通りです。

エージェント間連携の簡易さ
Google Cloud統合
MCP Toolbox for Databasesによる簡単なBigQuery接続

エージェント間連携の簡易さ

ADKでは、次のようにシンプルなコードでエージェント間でのオーケストレーションを実装できます。

今回のケースでは、企業の業務情報や従業員数をWEBから取得したいユースケースがあったので、Google検索ができるエージェントをサブエージェントとして実装しています。

# Google検索で企業の情報を取得するエージェント
google_search_agent = LlmAgent(
    model="gemini-2.5-flash",
    name="google_search_agent",
    description="A helpful AI assistant that can search company information from google.",
    instruction=GOOGLE_SEARCH_AGENT_INSTR,
    tools=[google_search],
)

root_agent = LlmAgent(
    model="gemini-2.5-pro",
    name="company_master_agent",
    description="A helpful AI assistant that can search company.",
    instruction=ROOT_AGENT_INSTR,
    sub_agents=[google_search_agent] #Google検索エージェントをサブエージェントとして追加
)

タスクごとに細かくサブエージェントを立てられるため、非常に見通しの良いエージェントシステムが構築できます。

Google Cloud統合

データ基盤をBigQueryで構築しており、Geminiも活用している弊社にとって、権限やホスティングに必要な知識を新しくキャッチアップする必要がないのはかなり大きかったです。

MCP Toolbox for Databasesによる簡単なBigQuery接続

MCP Toolbox for Databasesでは、次のようなyamlファイルを定義するだけで簡単にBigQueryへの接続情報をMCP化できます。

my-bigquery-source:
  kind: bigquery
  project: ${your_project}
  location: asia-northeast1
toolsets:
  search_company:
    - search-company-by-name
tools:
  search-company-by-name:
    kind: bigquery-sql
    source: my-bigquery-source
    description: 企業情報を企業名から検索する。
    parameters:
      - name: company_name
        type: string
        description: 企業名
    statement: SELECT * FROM `${your_dataset}.companies` WHERE company_name LIKE CONCAT('%', @company_name,'%');

MCPサーバーは、次のようなコマンドで簡単に起動できます。

containers:
  - image: us-central1-docker.pkg.dev/database-toolbox/toolbox/toolbox:latest
    name: toolbox
    args:
      - "toolbox"
      - "--tools-file"
      - "/app/tools.yml"
      - "--address"
      - "0.0.0.0"
      - "--port"
      - "5001"

エージェントからは次のように参照させると、yaml上で登録したtool(今回はsearch-company-by-name)を使えるようになります。

from toolbox_core import ToolboxSyncClient

toolbox = ToolboxSyncClient("http://localhost:5001")
search_company_tools = toolbox.load_toolset("search_company")

agent = LlmAgent(
    model="gemini-2.5-pro",
    name="company_search_agent",
    description="A helpful AI assistant that can search company in BigQuery",
    tools=search_company_tools
)

このように事前にBigQueryとの接続を定義しておくことで、意図したSQLを使ってデータにアクセスさせる事ができ、不要なデータベースの参照や誤ったコマンドを防ぐことができました。

工夫した点

オーケストレーションの設計

メインエージェントのInstructionはサブエージェントのユースケースのみを記載し、BigQueryやGoogle検索の具体的な利用方法はサブエージェントのInstructionに記載しています。これにより、メインエージェントは全体のオーケストレーションに集中し、各サブエージェントは専門的なタスクに特化できるような設計となっています。

ツールを搭載したサブエージェントの呼び出しについて

2025年7月現在、ビルトインツールやPythonの関数で定義したツールを搭載したエージェントを2つ以上同時にサブエージェントとして呼び出すことができません。

github.com

エラー回避策として、エージェントをToolとして呼び出すAgentToolというラッパーを使っています。

# toolを持ったサブエージェントを定義
google_search_agent = LlmAgent(
  ~~~,
  tools=[google_search],
)
bigquery_search_agent = LlmAgent(
  ~~~,
  tools=[bigquery_search],
)
# AgentToolでラップ
google_search_agent_tool = AgentTool(agent=google_search_agent)
bigquery_search_agent_tool = AgentTool(agent=bigquery_search_agent)
# メインエージェントからはツールとして参照
root_agent = LlmAgent(
  ~~~,
  tools = [google_search_agent_tool, bigquery_search_agent_tool],
)

Slack統合

新しいツールを導入することで利用者の負担を増やすことを避けるため、既に社内で日常的に使用されているSlack上でやりとりできるシステムを構築しました。ソケットモードで送信されてきたメッセージでエージェントを起動し、返答をSlackに返却するMCPクライアントを実装しています。

class SlackToADKClient:
    """Slack メッセージを ADK エージェントに橋渡しする MCP クライアント"""

    def __init__(self):
        self.slack_app = AsyncApp(token=SLACK_BOT_TOKEN)
        self.session_service = InMemorySessionService()

    async def handle_slack_message(self, event, say):
        """Slackメッセージイベントを処理"""
        user_id = event.get("user")
        message_text = event.get("text", "")
        
        # エージェントに問い合わせ
        response = await self.query_adk_agent(message_text, user_id)
        await say(text=response)

    async def query_adk_agent(self, message: str, user_id: str):
        """エージェントにクエリを送信"""
        # セッション管理
        session = await self.session_service.get_or_create_session(~~~)

        runner = Runner(
            agent=self.company_agent,
            session_service=self.session_service,
        )
        # エージェントの実行
        events = runner.run_async(~~~)
        # ~~~応答処理~~~
        return response_text