Source code for tools.discord_moderation

"""Discord moderation actions (kick, ban, timeout, block, nickname)."""

from __future__ import annotations

import datetime
import logging
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from tool_context import ToolContext

logger = logging.getLogger(__name__)

TOOL_NAME = "discord_moderation"
TOOL_DESCRIPTION = (
    "Perform moderation actions in a Discord server. "
    "Actions: 'kick', 'ban', 'timeout', 'block', 'unblock', "
    "'change_nickname'. Requires admin permissions."
)
TOOL_PARAMETERS = {
    "type": "object",
    "properties": {
        "action": {
            "type": "string",
            "enum": [
                "kick", "ban", "timeout",
                "block", "unblock", "change_nickname",
            ],
            "description": "The moderation action.",
        },
        "server_id": {
            "type": "string",
            "description": "Server (guild) ID.",
        },
        "target_user_id": {
            "type": "string",
            "description": "User to moderate.",
        },
        "reason": {
            "type": "string",
            "description": "Audit log reason.",
        },
        "duration_minutes": {
            "type": "integer",
            "description": (
                "Timeout duration in minutes (timeout only, "
                "max 40320 = 28 days)."
            ),
        },
        "nickname": {
            "type": "string",
            "description": (
                "New nickname (change_nickname only). "
                "Use empty string to reset."
            ),
        },
        "channel_id": {
            "type": "string",
            "description": (
                "Channel ID (for block/unblock overrides)."
            ),
        },
    },
    "required": ["action", "server_id", "target_user_id"],
}


[docs] async def run( action: str, server_id: str, target_user_id: str, reason: str | None = None, duration_minutes: int = 10, nickname: str | None = None, channel_id: str | None = None, ctx: ToolContext | None = None, ) -> str: """Execute this tool and return the result. Args: action (str): The action value. server_id (str): The server id value. target_user_id (str): The target user id value. reason (str | None): The reason value. duration_minutes (int): The duration minutes value. nickname (str | None): The nickname value. channel_id (str | None): Discord/Matrix channel identifier. ctx (ToolContext | None): Tool execution context providing access to bot internals. Returns: str: Result string. """ import discord from tools._discord_helpers import ( require_discord_client, check_admin_permission, ) client = require_discord_client(ctx) if isinstance(client, str): return client user_id = ctx.user_id if ctx else "" ok, err = await check_admin_permission( client, user_id, server_id, ) if not ok: return err try: server = client.get_guild(int(server_id)) except ValueError: return f"Error: Invalid server ID: '{server_id}'." if not server: return f"Error: Server '{server_id}' not found." try: target = server.get_member(int(target_user_id)) except ValueError: return ( f"Error: Invalid user ID: '{target_user_id}'." ) if not target and action != "ban": return ( f"Error: User '{target_user_id}' not found in " f"'{server.name}'." ) audit_reason = reason or f"Action by bot (requested by {user_id})" # --- KICK --- if action == "kick": try: await target.kick(reason=audit_reason) return ( f"Kicked '{target.name}' from '{server.name}'." ) except discord.errors.Forbidden: return "Error: No permission to kick this user." except Exception as exc: return f"Error kicking user: {exc}" # --- BAN --- if action == "ban": try: if target: await target.ban(reason=audit_reason) return ( f"Banned '{target.name}' from " f"'{server.name}'." ) else: user = discord.Object(id=int(target_user_id)) await server.ban(user, reason=audit_reason) return ( f"Banned user '{target_user_id}' from " f"'{server.name}'." ) except discord.errors.Forbidden: return "Error: No permission to ban this user." except Exception as exc: return f"Error banning user: {exc}" # --- TIMEOUT --- if action == "timeout": duration_minutes = max(1, min(duration_minutes, 40320)) until = ( datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=duration_minutes) ) try: await target.timeout(until, reason=audit_reason) return ( f"Timed out '{target.name}' for " f"{duration_minutes} minutes." ) except discord.errors.Forbidden: return "Error: No permission to timeout this user." except Exception as exc: return f"Error timing out user: {exc}" # --- BLOCK --- if action == "block": cid = channel_id or (ctx.channel_id if ctx else "") if not cid: return "Error: channel_id is required for block." try: ch = client.get_channel(int(cid)) except ValueError: return f"Error: Invalid channel ID: '{cid}'." if not ch or not hasattr(ch, "set_permissions"): return f"Error: Channel '{cid}' not found." try: overwrite = ch.overwrites_for(target) overwrite.send_messages = False overwrite.add_reactions = False await ch.set_permissions( target, overwrite=overwrite, reason=audit_reason, ) ch_name = getattr(ch, "name", cid) return ( f"Blocked '{target.name}' in channel " f"'{ch_name}'." ) except discord.errors.Forbidden: return "Error: No permission to set channel perms." except Exception as exc: return f"Error blocking user: {exc}" # --- UNBLOCK --- if action == "unblock": cid = channel_id or (ctx.channel_id if ctx else "") if not cid: return "Error: channel_id is required for unblock." try: ch = client.get_channel(int(cid)) except ValueError: return f"Error: Invalid channel ID: '{cid}'." if not ch or not hasattr(ch, "set_permissions"): return f"Error: Channel '{cid}' not found." try: overwrite = ch.overwrites_for(target) overwrite.send_messages = None overwrite.add_reactions = None await ch.set_permissions( target, overwrite=overwrite, reason=audit_reason, ) ch_name = getattr(ch, "name", cid) return ( f"Unblocked '{target.name}' in channel " f"'{ch_name}'." ) except discord.errors.Forbidden: return "Error: No permission to set channel perms." except Exception as exc: return f"Error unblocking user: {exc}" # --- CHANGE NICKNAME --- if action == "change_nickname": new_nick = nickname if new_nick is None: return "Error: 'nickname' is required." new_nick = new_nick or None # empty string -> reset try: await target.edit( nick=new_nick, reason=audit_reason, ) display = new_nick if new_nick else "(reset)" return ( f"Changed '{target.name}' nickname to " f"'{display}' in '{server.name}'." ) except discord.errors.Forbidden: return "Error: No permission to change nickname." except Exception as exc: return f"Error changing nickname: {exc}" return f"Error: Unknown action '{action}'."