【MCP】AutoGenとMCPによる自律エージェント連携：Googleカレンダー・社内DB・Notionをつなぐ実装例　2025/10/8

概要

この記事では、AutoGenのSelectorGroupChat機能を活用し、
LLMエージェントが自律的に判断して外部ツール（MCPサーバー経由）を操作する仕組みを紹介する。
対象となるMCPサーバーは以下の3つである。

Google Calendar MCP
社内DB MCP
Notion MCP

目的は、ユーザーが「来週の打ち合わせデータをまとめて」と指示するだけで、
エージェントが自動的に情報を取得・整理・記録するシステムを構築することだ。

システム構成

User ──▶ AutoGen SelectorGroupChat
           ├── CalendarAgent ──▶ MCP: Google Calendar
           ├── DBAgent       ──▶ MCP: internal-db
           └── NotionAgent   ──▶ MCP: Notion

エージェント間の発話順序（どのエージェントが次に動くか）は、
candidate_funcとselector_funcによって自律的に決定される。

実装コード例

# pip install pyautogen mcp
# AutoGenとMCPクライアントをインストール

from autogen import AssistantAgent, UserProxyAgent, SelectorGroupChat, GroupChatManager
from mcp import Client
# AutoGen主要クラスとMCPクライアントをインポート

# --- 各エージェント定義 ---
# Google Calendar、DB、Notionの3エージェントを定義

class CalendarAgent(AssistantAgent):
    # Googleカレンダー操作エージェント
    def __init__(self):
        super().__init__(name="CalendarAgent",
                         system_message="You operate Google Calendar via MCP.")
        self.client = Client("mcp://google-calendar")  # GoogleカレンダーMCPに接続

    def run(self, msg):
        # 「来週の予定」を取得
        events = self.client.call("get_events", {"time_range": "next_week"})
        return events


class DBAgent(AssistantAgent):
    # 社内DBを検索するエージェント
    def __init__(self):
        super().__init__(name="DBAgent",
                         system_message="You query internal DB via MCP.")
        self.client = Client("mcp://internal-db")  # 内部DB MCPに接続

    def run(self, events):
        # イベントごとにDBから関連データを検索
        bundles = []
        for e in events:
            data = self.client.call("search_meeting_data", {"keyword": e["title"]})
            bundles.append({"event": e, "data": data})
        return bundles


class NotionAgent(AssistantAgent):
    # Notionへ書き込むエージェント
    def __init__(self):
        super().__init__(name="NotionAgent",
                         system_message="You write results to Notion via MCP.")
        self.client = Client("mcp://notion")  # Notion MCPに接続

    def run(self, bundles):
        # Notionにページを作成
        result = self.client.call("create_page", {
            "title": "来週の打ち合わせまとめ",
            "content": str(bundles)
        })
        return result


# --- インスタンス生成 ---
user = UserProxyAgent(name="User")
calendar_agent = CalendarAgent()
db_agent = DBAgent()
notion_agent = NotionAgent()


# --- candidate_func ---
# 履歴に基づき、次に話す候補エージェントを選定
def candidate_func(history):
    text = " ".join([m.get("content", "") for m in history])
    cands = []
    if "予定" in text or "来週" in text or "calendar" in text:
        cands.append(calendar_agent)
    if "CalendarOK" in text or "events:" in text:
        cands.append(db_agent)
    if "DBOK" in text or "bundles:" in text:
        cands.append(notion_agent)
    return cands or [calendar_agent]


# --- selector_func ---
# 候補の中から実際に発話するエージェントを選択
def selector_func(candidates, history):
    text = " ".join([m.get("content", "") for m in history])
    if any(k in text for k in ["bundles:", "DBOK"]) and notion_agent in candidates:
        return notion_agent
    if any(k in text for k in ["events:", "CalendarOK"]) and db_agent in candidates:
        return db_agent
    return candidates[0]


# --- SelectorGroupChat設定 ---
groupchat = SelectorGroupChat(
    agents=[user, calendar_agent, db_agent, notion_agent],
    messages=[],
    max_round=6,
    candidate_func=candidate_func,
    selector_func=selector_func,
)
manager = GroupChatManager(groupchat=groupchat)


# --- 会話シミュレーション ---
user_message = "来週の打ち合わせデータをまとめて"
groupchat.append(message=user_message, role="user")

for _ in range(6):
    agent = groupchat.select_speaker()

    if agent is calendar_agent:
        # カレンダーから予定取得
        events = calendar_agent.run(user_message)
        groupchat.append(role="assistant", name=calendar_agent.name,
                         message=f"CalendarOK events:{events}")

    elif agent is db_agent:
        # イベントを基にDB検索
        bundles = db_agent.run(events)
        groupchat.append(role="assistant", name=db_agent.name,
                         message=f"DBOK bundles:{bundles}")

    elif agent is notion_agent:
        # Notionにまとめを書き込み
        result = notion_agent.run(bundles)
        groupchat.append(role="assistant", name=notion_agent.name,
                         message=f"NotionOK {result}")
        break


# --- 出力 ---
# 会話履歴をすべて表示
for msg in groupchat.messages:
    print(f"{msg['role']} ({msg.get('name','')}): {msg['content']}")

実行結果例

user (User): 来週の打ち合わせデータをまとめて
assistant (CalendarAgent): CalendarOK events:[{'title': 'Design Review', 'time': '2025-10-15 10:00'}]
assistant (DBAgent): DBOK bundles:[{'event': {...}, 'data': ['spec_v3.pdf', 'kpi.xlsx']}]
assistant (NotionAgent): NotionOK {'page_url': 'https://notion.so/meeting-summary-2025'}

処理フローの解説

ユーザーが依頼を送信
AutoGenのUserProxyAgentが初期発話を登録。
candidate_funcが候補を生成
会話履歴から「予定」などのキーワードを解析し、CalendarAgentを候補に。
selector_funcが担当エージェントを決定
状態に応じてCalendar → DB → Notionの順で自律的に切り替え。
各エージェントがMCPサーバーを呼び出し
Client("mcp://service-name") を通じ、実際のMCP API（Google Calendar、Notionなど）へアクセス。
結果をNotionに出力して完了

技術的ポイント

SelectorGroupChat
AutoGenの中核。エージェント選択を自律化できる。
candidate_func / selector_func
ルールベースやLLM出力により、次の発話者を動的決定。
MCPプロトコル
外部ツール呼び出しを標準化。
例：mcp://google-calendar はGoogle Calendar APIの抽象化レイヤー。
再利用性
MCPサーバーを入れ替えれば、SlackやGitHubなど他の外部ツールとも即統合可能。