"""Send a Direct Message to a Discord user."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "discord_send_dm"
TOOL_DESCRIPTION = (
"Sends a Direct Message (DM) to a specified Discord user. "
"Useful for private notifications regardless of the current channel."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"target_user_id": {
"type": "string",
"description": "The Discord user ID to send the DM to.",
},
"message": {
"type": "string",
"description": "The message content to send.",
},
},
"required": ["target_user_id", "message"],
}
[docs]
async def run(
target_user_id: str,
message: str,
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
target_user_id (str): The target user id value.
message (str): The message value.
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
client = require_discord_client(ctx)
if isinstance(client, str):
return client
caller = (getattr(ctx, "user_id", "") or "").strip()
dst = (target_user_id or "").strip()
if dst and dst != caller:
redis = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
if redis is None or config is None:
return (
"Error: Sending a DM to another user requires Redis and "
"config for privilege checks."
)
from tools.alter_privileges import has_privilege, PRIVILEGES
if not await has_privilege(redis, caller, PRIVILEGES["READ_DM"], config):
return (
"Error: Sending a DM to another user requires the READ_DM "
"privilege (admins implicitly have it)."
)
try:
user = client.get_user(int(target_user_id))
if not user:
try:
user = await client.fetch_user(int(target_user_id))
except discord.errors.NotFound:
return (
f"Error: User with ID '{target_user_id}' "
f"not found."
)
except discord.errors.Forbidden:
return (
f"Error: I don't have permission to access "
f"user '{target_user_id}'."
)
except Exception as exc:
return (
f"Error: Failed to fetch user "
f"'{target_user_id}': {exc}"
)
if not user:
return (
f"Error: User with ID '{target_user_id}' not found."
)
dm_channel = await user.create_dm()
await dm_channel.send(message)
preview = message[:50] + ("..." if len(message) > 50 else "")
return (
f"Successfully sent DM to user '{user.name}' "
f"with message: '{preview}'"
)
except discord.errors.Forbidden:
return (
"Error: I don't have permission to send DMs to that "
"user. They may have DMs disabled."
)
except ValueError:
return (
f"Error: Invalid user ID format: '{target_user_id}'. "
f"User ID must be a number."
)
except Exception as exc:
return f"An unexpected error occurred: {exc}"