#!/usr/bin/env python3 """ Discord Agent Bot AI-powered Discord bot using Ollama. Configuration loaded from ./config/agent-config.yaml """ import discord from discord.ext import commands import os import asyncio import aiohttp import logging import yaml import re from typing import Any, Dict, Optional from datetime import datetime # ------------------------- # Constants # ------------------------- CONFIG_PATH = "config/agent-config.yaml" # ------------------------- # Configuration helpers # ------------------------- def load_config(path: str) -> Dict[str, Any]: """Load YAML config with environment variable substitution.""" def substitute(value: Any) -> Any: if isinstance(value, dict): return {k: substitute(v) for k, v in value.items()} if isinstance(value, list): return [substitute(v) for v in value] if isinstance(value, str): pattern = r"\$\{([^}:]+)(?::([^}]*))?\}" return re.sub( pattern, lambda m: os.getenv(m.group(1), m.group(2) or ""), value, ) return value try: with open(path, "r") as f: raw = yaml.safe_load(f) or {} config = substitute(raw) config["_config_path"] = path logging.info(f"Loaded config from {path}") return config except FileNotFoundError: logging.error(f"Config file not found: {path}") except Exception as e: logging.error(f"Failed to load config: {e}") return { "_config_path": path, "discord": {"token": "", "prefix": ";"}, "ollama": {"endpoint": "http://192.168.0.31:11434", "model": "ministral-3:8b"}, "agent": {"instructions": "You are a helpful AI assistant."}, "features": {"max_response_length": 1900}, "advanced": {"admin_users": []}, } def cfg(config: Dict[str, Any], path: str, default=None): cur = config for key in path.split("."): if not isinstance(cur, dict) or key not in cur: return default cur = cur[key] return cur # ------------------------- # Discord bot # ------------------------- class DiscordAgent(commands.Bot): def __init__(self, config_path: str): self.config_path = config_path self.config = load_config(config_path) intents = discord.Intents.default() intents.message_content = True super().__init__( command_prefix=cfg(self.config, "discord.prefix", ";"), intents=intents, # help_command=None, ) self.session: Optional[aiohttp.ClientSession] = None self.apply_config() async def on_message(self, message): if message.author == self.user: return print(f"Message seen: {message.content}") # Debug log await self.process_commands(message) # Necessary to make commands work def apply_config(self): """Apply configuration to runtime attributes.""" self.ollama_endpoint = cfg(self.config, "ollama.endpoint", "http://192.168.0.31:11434") self.ollama_model = cfg(self.config, "ollama.model", "ministral-3:8b") self.ollama_params = cfg(self.config, "ollama.parameters", {}) self.ollama_timeout = cfg(self.config, "ollama.timeout", 60) self.agent_instructions = cfg( self.config, "agent.instructions", "You are a helpful AI assistant.", ) self.max_response_length = cfg(self.config, "features.max_response_length", 1900) async def setup_hook(self): self.session = aiohttp.ClientSession() async def close(self): if self.session: await self.session.close() await super().close() async def on_ready(self): activity = discord.Activity( type=discord.ActivityType.watching, name=cfg(self.config, "discord.status", "AI Assistant"), ) await self.change_presence(activity=activity) logging.info(f"Logged in as {self.user} ({self.user.id})") print(f"✅ Bot ready as {self.user}") async def call_ollama(self, message: str) -> str: payload = { "model": self.ollama_model, "messages": [ {"role": "system", "content": self.agent_instructions}, {"role": "user", "content": message}, ], "stream": False, "options": self.ollama_params, } try: async with self.session.post( f"{self.ollama_endpoint}/api/chat", json=payload, timeout=aiohttp.ClientTimeout(total=self.ollama_timeout), ) as r: if r.status == 200: data = await r.json() return data.get("message", {}).get("content", "") return f"❌ Ollama HTTP {r.status}" except asyncio.TimeoutError: return "❌ Ollama request timed out" except Exception as e: logging.error(f"Ollama error: {e}") return f"❌ Ollama error: {e}" # ------------------------- # Commands # ------------------------- bot = DiscordAgent(CONFIG_PATH) @bot.command() async def agent(ctx, *, message: str): """Chat with the AI agent.""" async with ctx.typing(): reply = await bot.call_ollama(message) await ctx.reply(reply[: bot.max_response_length]) @bot.command() async def reload(ctx): """Reload configuration (admin only).""" admins = cfg(bot.config, "advanced.admin_users", []) if admins and ctx.author.id not in admins: await ctx.reply("❌ You don't have permission to do that.") return bot.config = load_config(bot.config_path) bot.apply_config() await ctx.reply("✅ Configuration reloaded") logging.info(f"Config reloaded by {ctx.author}") @bot.command() async def status(ctx): """Show bot status.""" embed = discord.Embed( title="📊 Bot Status", color=discord.Color.green(), timestamp=datetime.utcnow(), ) embed.add_field(name="Model", value=bot.ollama_model) embed.add_field(name="Latency", value=f"{round(bot.latency * 1000)}ms") embed.add_field(name="Config", value=bot.config_path) await ctx.reply(embed=embed) @bot.event async def on_command_error(ctx, error): if isinstance(error, commands.CommandNotFound): return logging.error(f"Command error: {error}") await ctx.reply("❌ An error occurred while processing that command") # ------------------------- # Entrypoint # ------------------------- def main(): logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) token = cfg(bot.config, "discord.token") if not token: print("❌ DISCORD_BOT_TOKEN not set") return 1 try: bot.run(token, log_handler=None) except KeyboardInterrupt: print("\n👋 Shutting down") return 0 if __name__ == "__main__": raise SystemExit(main())