
Agent Development KitでDiscord用エージェントを作成する
2025年05月19日
ADKDiscord
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,
)