ClineでとりあえずDiscordエージェント動かすところまで作ったメモ。

リファクタリング一切してない前提（他のコードコピーしていじってるので不要な部分が結構あり）。 コーディネータから呼び出される前提。

zapier mcp使うとめっちゃ楽にできるが無料枠が少ないのでツール実装する方式。

事前準備

Discord用のトークンは取得しておく（Web検索すると色々出てくる）

Discord tool用のプログラム

import os
import sys
import asyncio
from typing import List, Dict, Optional, Any
import discord
from discord.ext import commands

# パラメータとサブモジュールimport
parent_dir = os.path.dirname(os.path.abspath(__file__))
root_dir = os.path.dirname(parent_dir)
sys.path.append(parent_dir)
sys.path.append(root_dir)
from shared_libraries import constants

# Discordクライアントのセットアップ
intents = discord.Intents.default()
intents.message_content = True  # メッセージ内容へのアクセス権限
client = commands.Bot(command_prefix='!', intents=intents)

# Discordトークンの設定（実際の使用時は.envファイルやシークレット管理を推奨）
DISCORD_TOKEN = constants.DISCORD_TOKEN

# クライアントの状態を追跡
client_initialized = False
client_ready = asyncio.Event()

# ====== Discord関連の関数 ======

@client.event
async def on_ready():
    """クライアントが準備完了したときに呼ばれるイベント"""
    global client_ready
    print(f'Logged in as {client.user} (ID: {client.user.id})')
    client_ready.set()  # イベントをセット

async def initialize_discord_client():
    """Discordクライアントを初期化して接続します"""
    global client_initialized
    
    if not client_initialized:
        # バックグラウンドでクライアントを起動
        asyncio.create_task(client.start(DISCORD_TOKEN))
        
        # クライアントが準備完了するまで待機（タイムアウトは30秒）
        try:
            await asyncio.wait_for(client_ready.wait(), timeout=30.0)
            client_initialized = True
            return {"status": "success", "message": "Discordクライアントが初期化されました"}
        except asyncio.TimeoutError:
            return {"status": "error", "error_message": "Discordクライアントの初期化がタイムアウトしました"}
    else:
        return {"status": "success", "message": "Discordクライアントは既に初期化されています"}

async def search_messages(channel_id: str, query: str, limit: int = 100) -> Dict[str, Any]:
    """
    指定されたチャンネル内でメッセージを検索します。
    
    Args:
        channel_id (str): 検索対象のDiscordチャンネルID
        query (str): 検索クエリ（メッセージ内容に含まれる文字列）
        limit (int, optional): 取得するメッセージ数の上限。デフォルトは100。
    
    Returns:
        Dict[str, Any]: 検索結果と状態を含む辞書
    """
    # クライアントが初期化されているか確認
    if not client_initialized:
        return {
            "status": "error",
            "error_message": "Discordクライアントが初期化されていません。先にdiscord_init()を実行してください。"
        }
    
    try:
        # チャンネルを取得
        channel = client.get_channel(int(channel_id))
        if channel is None:
            return {
                "status": "error",
                "error_message": f"チャンネルID {channel_id} が見つかりませんでした。"
            }
        
        # メッセージを取得して検索
        messages = []
        async for message in channel.history(limit=limit):
            if query.lower() in message.content.lower():
                messages.append({
                    "id": message.id,
                    "author": str(message.author),
                    "content": message.content,
                    "created_at": message.created_at.isoformat(),
                    "jump_url": message.jump_url
                })
        
        return {
            "status": "success",
            "channel_name": channel.name,
            "message_count": len(messages),
            "messages": messages
        }
    
    except Exception as e:
        return {
            "status": "error",
            "error_message": f"メッセージの検索中にエラーが発生しました: {str(e)}"
        }

async def send_message(channel_id: str, content: str) -> Dict[str, Any]:
    """
    指定されたチャンネルにメッセージを送信します。
    
    Args:
        channel_id (str): 送信先のDiscordチャンネルID
        content (str): 送信するメッセージ内容
    
    Returns:
        Dict[str, Any]: 送信結果と状態を含む辞書
    """
    # クライアントが初期化されているか確認
    if not client_initialized:
        return {
            "status": "error",
            "error_message": "Discordクライアントが初期化されていません。先にdiscord_init()を実行してください。"
        }
    
    try:
        # チャンネルを取得
        channel = client.get_channel(int(channel_id))
        if channel is None:
            return {
                "status": "error",
                "error_message": f"チャンネルID {channel_id} が見つかりませんでした。"
            }
        
        # メッセージを送信
        message = await channel.send(content)
        
        return {
            "status": "success",
            "message_id": message.id,
            "channel_name": channel.name,
            "content": content
        }
    
    except Exception as e:
        return {
            "status": "error",
            "error_message": f"メッセージの送信中にエラーが発生しました: {str(e)}"
        }

async def get_channel_list() -> Dict[str, Any]:
    """
    アクセス可能なチャンネルの一覧を取得します。
    
    Returns:
        Dict[str, Any]: チャンネル一覧と状態を含む辞書
    """
    # クライアントが初期化されているか確認
    if not client_initialized:
        return {
            "status": "error",
            "error_message": "Discordクライアントが初期化されていません。先にdiscord_init()を実行してください。"
        }
    
    try:
        channels = []
        for guild in client.guilds:
            for channel in guild.text_channels:
                channels.append({
                    "id": channel.id,
                    "name": channel.name,
                    "guild_name": guild.name,
                    "guild_id": guild.id
                })
        
        return {
            "status": "success",
            "channel_count": len(channels),
            "channels": channels
        }
    
    except Exception as e:
        return {
            "status": "error",
            "error_message": f"チャンネル一覧の取得中にエラーが発生しました: {str(e)}"
        }

import nest_asyncio
nest_asyncio.apply()

async def main():
    await initialize_discord_client()

asyncio.run(main())

エージェントの方のプログラム

# Add nest_asyncio to allow running asyncio.run within an existing event loop
import nest_asyncio
nest_asyncio.apply()

import os
import sys
import asyncio
from dotenv import load_dotenv
from google.genai import types
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams # Added MCPToolset imports here

# discord_tool.pyからツール関数をインポート
from .discord_tool import (
    initialize_discord_client,
    search_messages,
    send_message,
    get_channel_list
)

load_dotenv('.env')

parent_dir = os.path.dirname(os.path.abspath(__file__))
root_dir = os.path.dirname(parent_dir)
sys.path.append(parent_dir)
sys.path.append(root_dir)
from shared_libraries import constants
from shared_libraries.error_handling import (
    before_agent_callback,
    after_agent_callback,
)
from . import prompt
#import prompt

AGENT_MODEL = constants.MODEL_GEMINI_2_5_PRO # Starting with a powerful Gemini model

# ====== ADKツールの定義 ======
# コルーチンを実行するヘルパー関数
def run_coroutine(coroutine):
    """コルーチンを実行してその結果を返す"""
    loop = asyncio.get_event_loop()
    return loop.run_until_complete(coroutine)

# 非同期関数をラップする同期関数
def discord_search(channel_id: str, query: str, limit: int = 100):
    """
    Discordチャンネル内のメッセージを検索するツール
    
    Args:
        channel_id (str): 検索対象のDiscordチャンネルID
        query (str): 検索クエリ（メッセージ内容に含まれる文字列）
        limit (int, optional): 取得するメッセージ数の上限。デフォルトは100。
    
    Returns:
        Dict[str, Any]: 検索結果と状態を含む辞書
    """
    return run_coroutine(search_messages(channel_id, query, limit))

def discord_send(channel_id: str, content: str):
    """
    Discordチャンネルにメッセージを送信するツール
    
    Args:
        channel_id (str): 送信先のDiscordチャンネルID
        content (str): 送信するメッセージ内容
    
    Returns:
        Dict[str, Any]: 送信結果と状態を含む辞書
    """
    return run_coroutine(send_message(channel_id, content))

def discord_channels():
    """
    アクセス可能なDiscordチャンネルの一覧を取得するツール
    
    Returns:
        Dict[str, Any]: チャンネル一覧と状態を含む辞書
    """
    return run_coroutine(get_channel_list())

def discord_init():
    """
    Discordクライアントを初期化するツール
    
    Returns:
        Dict[str, Any]: 初期化結果と状態を含む辞書
    """
    return run_coroutine(initialize_discord_client())

# ツールのインスタンス化（FunctionToolを使用）
discord_search_tool = FunctionTool(func=discord_search)
discord_send_tool = FunctionTool(func=discord_send)
discord_channels_tool = FunctionTool(func=discord_channels)
discord_init_tool = FunctionTool(func=discord_init)

discord_root_agent = Agent(
        model=AGENT_MODEL, # Adjust model name if needed
        name=prompt.AGENT_NAME,
        description=prompt.DESCRIPTION,
        instruction=prompt.PROMPT,
        tools=[
            discord_init_tool,
            discord_channels_tool, 
            discord_search_tool, 
            discord_send_tool
        ],
        before_agent_callback=before_agent_callback,
        after_agent_callback=after_agent_callback,
    )