Discord Bot に AI チャット機能を実装しようとしたとき、多くの開発者が直面する壁があります。「ConnectionError: timeout after 30 seconds」「401 Unauthorized - Invalid API key」「RateLimitError: Too many requests」。これらのエラーは、適切なエンドポイント設定とエラーハンドリング缺失が原因です。

本記事では、HolySheep AI を使用して、 Discord Bot に多輪対話と工具調用（Function Calling）を実装する方法を実践的に解説します。

前提条件とプロジェクト構成

まず、必要な環境を整えます。本記事のコードは discord.py と OpenAI SDK 互換クライアントを使用します。

# requirements.txt
discord.py>=2.3.0
openai>=1.12.0
python-dotenv>=1.0.0
aiohttp>=3.9.0
# インストールコマンド
pip install -r requirements.txt

Step 1: 基本的セットアップ — 会話履歴なし

まずは最もシンプルな実装から始めましょう。HolySheep AI は OpenAI API 互換エンドポイントを 제공하는ため、openai Python SDK をそのまま流用できます。

# bot.py
import os
import discord
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()


HolySheep AI エンドポイント設定（絶対に api.openai.com は使用しない）

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # ここ重要！
)

DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")

intents = discord.Intents.default()
intents.message_content = True
bot = discord.Client(intents=intents)

@bot.event
async def on_message(message):
    # Bot 自身のメッセージは無視
    if message.author.bot:
        return
    
    # チャンネルでのみ応答
    if not isinstance(message.channel, discord.DMChannel):
        if not message.content.startswith("!ai "):
            return
    
    user_message = message.content.replace("!ai ", "").strip()
    
    try:
        async with message.channel.typing():
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "user", "content": user_message}
                ],
                max_tokens=1000,
                temperature=0.7
            )
            
            ai_reply = response.choices[0].message.content
            await message.reply(ai_reply)
            
    except Exception as e:
        await message.reply(f"エラーが発生しました: {type(e).__name__}: {str(e)}")

bot.run(DISCORD_TOKEN)

筆者の実践経験では、この基本形で動かす際最も多い失敗が base_url の設定です。「api.openai.com」と誤って書くと、2024年後半から急に API キーが無効化される事例が増えました。HolySheep AI の場合は必ず https://api.holysheep.ai/v1 を指定してください。

Step 2: 多輪対話を実現 — 会話履歴の管理

単発の質問応答だけでは AI Bot の可能性は半分です。多輪対話を実現するには、ユーザーごとに会話履歴を管理する必要があります。Discord では Guild（サーバー）ID と Channel ID 、そしてユーザー ID を組み合わせて一意のセッションを識別します。

# conversation_manager.py
import os
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime, timedelta

@dataclass
class ConversationHistory:
    """ユーザーごとの会話履歴を管理"""
    messages: list = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    last_interaction: datetime = field(default_factory=datetime.now)
    
    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        self.last_interaction = datetime.now()
    
    def get_messages(self) -> list:
        """古すぎるメッセージを自動削除（コスト最適化）"""
        cutoff = datetime.now() - timedelta(hours=1)
        if self.created_at < cutoff:
            # システムプロンプトのみ保持
            return [m for m in self.messages if m["role"] == "system"]
        return self.messages
    
    def clear(self):
        self.messages = []


class ConversationManager:
    """全ユーザーの会話を一元管理"""
    
    def __init__(self, max_history_per_user: int = 20):
        self.conversations: dict[str, ConversationHistory] = {}
        self.max_history = max_history_per_user
    
    def _get_session_key(self, guild_id: int, channel_id: int, user_id: int) -> str:
        """サーバー×チャンネル×ユーザー で一意のセッションキーを生成"""
        return f"{guild_id}:{channel_id}:{user_id}"
    
    def get_or_create(self, guild_id: int, channel_id: int, user_id: int) -> ConversationHistory:
        key = self._get_session_key(guild_id, channel_id, user_id)
        if key not in self.conversations:
            self.conversations[key] = ConversationHistory()
        return self.conversations[key]
    
    def add_user_message(self, guild_id: int, channel_id: int, 
                         user_id: int, content: str):
        conv = self.get_or_create(guild_id, channel_id, user_id)
        conv.add_message("user", content)
        
        # 履歴サイズ制限
        if len(conv.messages) > self.max_history:
            # system を除いて古いものを削除
            non_system = [m for m in conv.messages if m["role"] != "system"]
            system = [m for m in conv.messages if m["role"] == "system"]
            conv.messages = system + non_system[-(self.max_history-1):]
    
    def add_ai_message(self, guild_id: int, channel_id: int, 
                       user_id: int, content: str):
        conv = self.get_or_create(guild_id, channel_id, user_id)
        conv.add_message("assistant", content)
    
    def clear_conversation(self, guild_id: int, channel_id: int, user_id: int):
        key = self._get_session_key(guild_id, channel_id, user_id)
        if key in self.conversations:
            self.conversations[key].clear()



conversation_manager.py グローバルインスタンス

conv_manager = ConversationManager(max_history_per_user=15)

筆者が実際に運用していて気づいた点は、会話履歴のクリーンアップを怠ると API コストが跳ね上がることです。上記の実装では1時間経過した会話を自動リセットする仕組みを入れています。HolySheep AI の料金は ¥1=$1 と非常に安価ですが、それでも履歴管理は良好的なコスト控制に重要です。

# bot_advanced.py
import os
import discord
from dotenv import load_dotenv
from openai import OpenAI
from conversation_manager import conv_manager

load_dotenv()

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")

SYSTEM_PROMPT = """あなたは親しみやすいDiscordアシスタントです。
日本語で答え、簡潔で有用な情報を提供してください。
必要に応じてコードブロックやリストを使用して情報を整理してください。"""

intents = discord.Intents.default()
intents.message_content = True
bot = discord.Client(intents=intents)

@bot.event
async def on_message(message):
    if message.author.bot:
        return
    
    # コマンド処理
    if message.content.startswith("!ai "):
        await handle_ai_command(message)
    elif message.content == "!reset":
        await handle_reset_command(message)
    elif message.content == "!help":
        await send_help(message)

async def handle_ai_command(message):
    user_message = message.content.replace("!ai ", "").strip()
    
    if not user_message:
        await message.reply("質問を入力してください。例: !ai Pythonのリスト内包表記を教えてください")
        return
    
    guild_id = message.guild.id if message.guild else 0
    channel_id = message.channel.id
    user_id = message.author.id
    
    try:
        async with message.channel.typing():
            # 会話履歴に追加
            conv_manager.add_user_message(guild_id, channel_id, user_id, user_message)
            
            # 現在の会話履歴を取得
            conv = conv_manager.get_or_create(guild_id, channel_id, user_id)
            
            # システムプロンプトがまだなければ追加
            if not any(m["role"] == "system" for m in conv.messages):
                conv.add_message("system", SYSTEM_PROMPT)
            
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=conv.messages,
                max_tokens=2000,
                temperature=0.7
            )
            
            ai_reply = response.choices[0].message.content
            conv_manager.add_ai_message(guild_id, channel_id, user_id, ai_reply)
            
            await message.reply(ai_reply)
            
    except Exception as e:
        error_msg = f"⚠️ エラー: {type(e).__name__}\n{str(e)}"
        await message.reply(error_msg)

async def handle_reset_command(message):
    guild_id = message.guild.id if message.guild else 0
    channel_id = message.channel.id
    user_id = message.author.id
    
    conv_manager.clear_conversation(guild_id, channel_id, user_id)
    await message.reply("✅ 会話をリセットしました。")

async def send_help(message):
    help_text = """**利用可能なコマンド:**
!ai [質問] - AIに質問する（会話履歴あり）
!reset - 会話をリセット
!help - このヘルプを表示"""
    await message.reply(help_text)

bot.run(DISCORD_TOKEN)

Step 3: Function Calling（工具調用）の実装

Function Calling は、Bot がユーザーの代わりに外部ツールを実行できる機能です。例えば、天気予報、計算、データベース検索などを AI から直接行えます。HolySheep AI は GPT-4o シリーズに対応しているため、信頼性の高い Function Calling を実現できます。

# functions.py
import json
from datetime import datetime
from typing import Callable


利用可能な関数の定義（OpenAI仕様）

AVAILABLE_FUNCTIONS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "現在時刻を取得します。引数は不要です。",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "数式を計算します。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "計算式（例: 2 + 2, 10 * 5, 2**10）"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "指定した都市の天気を取得します。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "都市名（日本語または英語）"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度の単位"
                    }
                },
                "required": ["city"]
            }
        }
    }
]


関数の実際の実装

def get_current_time() -> str:
    """現在時刻を返す"""
    now = datetime.now()
    return now.strftime("%Y年%m月%d日 %H:%M:%S")

def calculate(expression: str) -> str:
    """安全な計算を実行"""
    try:
        # eval は危険なので、許可する文字を制限
        allowed_chars = set("0123456789+-*/.()** ")
        if not all(c in allowed_chars for c in expression):
            return "エラー: 無効な文字が含まれています"
        
        result = eval(expression)
        return f"{expression} = {result}"
    except ZeroDivisionError:
        return "エラー: ゼロで割ることはできません"
    except Exception as e:
        return f"計算エラー: {str(e)}"

def get_weather(city: str, unit: str = "celsius") -> str:
    """モック天気を返す（実際はAPI呼び出しに置き換え）"""
    # 実際には weatherapi.com などの外部APIを使用
    mock_data = {
        "東京": {"temp_c": 22, "condition": "晴れ", "humidity": 65},
        "大阪": {"temp_c": 24, "condition": "曇り", "humidity": 70},
        "ニューヨーク": {"temp_c": 18, "condition": "晴れ", "humidity": 55},
        "ロンドン": {"temp_c": 14, "condition": "雨", "humidity": 85}
    }
    
    if city in mock_data:
        data = mock_data[city]
        temp = data["temp_c"]
        if unit == "fahrenheit":
            temp = temp * 9/5 + 32
            unit_symbol = "°F"
        else:
            unit_symbol = "°C"
        return f"{city}: {data['condition']}, 気温: {temp}{unit_symbol}, 湿度: {data['humidity']}%"
    
    return f"{city} の天気データは利用できません"


関数名を実装にマッピング

FUNCTION_IMPLEMENTATIONS: dict[str, Callable] = {
    "get_current_time": get_current_time,
    "calculate": calculate,
    "get_weather": get_weather
}
# bot_function_calling.py
import os
import json
import discord
from dotenv import load_dotenv
from openai import OpenAI
from functions import AVAILABLE_FUNCTIONS, FUNCTION_IMPLEMENTATIONS

load_dotenv()

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")
MAX_FUNCTION_CALLS = 3  # 安全のための再帰呼び出し上限

intents = discord.Intents.default()
intents.message_content = True
bot = discord.Client(intents=intents)

async def execute_function_call(function_name: str, arguments: dict) -> str:
    """関数を実行して結果を返す"""
    if function_name not in FUNCTION_IMPLEMENTATIONS:
        return f"エラー: 未知の関数 {function_name}"
    
    try:
        func = FUNCTION_IMPLEMENTATIONS[function_name]
        result = func(**arguments)
        return str(result)
    except TypeError as e:
        return f"引数エラー: {str(e)}"
    except Exception as e:
        return f"関数実行エラー: {type(e).__name__}: {str(e)}"

@bot.event
async def on_message(message):
    if message.author.bot:
        return
    
    if not message.content.startswith("!func "):
        return
    
    user_message = message.content.replace("!func ", "").strip()
    
    if not user_message:
        examples = """**Function Calling の例:**
- !func 東京の天気を教えて
- !func 123の平方根を計算して
- !func 今何時？"""
        await message.reply(examples)
        return
    
    try:
        await message.channel.typing()
        
        # Function Calling 対応のモデルを使用
        response = client.chat.completions.create(
            model="gpt-4o-mini",  # Function Calling 対応モデル
            messages=[
                {"role": "user", "content": user_message}
            ],
            tools=AVAILABLE_FUNCTIONS,
            tool_choice="auto",
            max_tokens=1000
        )
        
        response_message = response.choices[0].message
        
        # 関数呼び出しがある場合
        if response_message.tool_calls:
            tool_results = []
            messages = [{"role": "user", "content": user_message}, response_message]
            
            for call in response_message.tool_calls[:MAX_FUNCTION_CALLS]:
                function_name = call.function.name
                arguments = json.loads(call.function.arguments)
                
                result = await execute_function_call(function_name, arguments)
                tool_results.append(f"📌 {function_name}: {result}")
                
                # 関数結果をmessagesに追加
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": result
                })
            
            # 関数結果をAIに解釈させる
            final_response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                max_tokens=1000
            )
            
            final_content = final_response.choices[0].message.content
            await message.reply(f"{chr(10).join(tool_results)}\n\n🤖 {final_content}")
        
        # 関数呼び出しがない場合
        else:
            await message.reply(response_message.content)
            
    except Exception as e:
        error_msg = f"⚠️ エラー: {type(e).__name__}\n{str(e)}"
        await message.reply(error_msg)

bot.run(DISCORD_TOKEN)

価格比較とコスト最適化

AI Bot の運用においてコストは見逃せない要素です。HolySheep AI は ¥1=$1 という業界最安水準の料金体系を提供します。以下は主要モデルの出力コスト比較です（2026年1月時点）：

筆者の实践经验では、Discord Bot には Gemini 2.5 Flash や DeepSeek V3.2 で十分な精度が得られます。GPT-4o-mini はコストと性能のバランスが非常に悪く、複雑な Reasoning が不要なら選択肢になりにくいです。

.env ファイルの設定

# .env

HolySheep AI API Key（https://www.holysheep.ai/register で取得）

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY


Discord Bot Token（Discord Developer Portal で作成）

DISCORD_TOKEN=your_discord_bot_token_here

よくあるエラーと対処法

1. 401 Unauthorized - Invalid API Key

# ❌ よくある間違い
client = OpenAI(
    api_key="sk-xxxxx",  # キー自体は有効だが
    base_url="https://api.openai.com/v1"  # エンドポイントが違う
)


✅ 正しい設定

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # HolySheep のエンドポイント
)

原因: base_url に api.openai.com を指定したままキーを入れ替えると、元のキーで OpenAI にアクセスしようとして失敗します。解決方法: base_url を必ず https://api.holysheep.ai/v1 に設定してください。API キーも HolySheep AI のダッシュボード で取得したものに変更する必要があります。

2. RateLimitError: Too Many Requests

# ❌ rate limit 考慮なし
async def handle_request(message):
    response = client.chat.completions.create(...)  # 同時大量呼び出し発生
    await message.reply(response)


✅ rate limit 対応

import asyncio
from collections import defaultdict
from time import time

class RateLimiter:
    def __init__(self, max_requests: int = 60, window: int = 60):
        self.max_requests = max_requests
        self.window = window
        self.requests = defaultdict(list)
    
    async def acquire(self, key: str):
        now = time()
        # ウィンドウ内の古いリクエストを削除
        self.requests[key] = [
            t for t in self.requests[key] 
            if now - t < self.window
        ]
        
        if len(self.requests[key]) >= self.max_requests:
            wait_time = self.window - (now - self.requests[key][0])
            raise RateLimitError(f"Rate limit. Wait {wait_time:.1f}s")
        
        self.requests[key].append(now)
        await asyncio.sleep(0.1)  # バッファ

rate_limiter = RateLimiter(max_requests=30, window=60)

async def handle_request_safe(message, limiter):
    await limiter.acquire(f"user_{message.author.id}")
    response = client.chat.completions.create(...)
    return response

原因: Discord は 동시에複数のメッセージを受け取るため、Bot が API に殺到して rate limit に抵触します。

関連リソース

関連記事