"""Discord webhook management and execution."""
from __future__ import annotations
import jsonutil as json
import logging
from typing import Any, TYPE_CHECKING
from urllib.parse import urlparse
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
def _validate_discord_webhook_url(webhook_url: str) -> str | None:
"""Return an error string if *webhook_url* is not a genuine Discord webhook URL.
SSRF guard: ``discord.Webhook.from_url`` performs the HTTP request *inside*
discord.py, so it cannot be routed through ``tools._safe_http``. Instead we
require the URL to be an HTTPS Discord endpoint, which structurally cannot
resolve to an internal host -- blocking a model-supplied target such as
``http://10.10.0.5:6379/`` that ``execute`` would otherwise POST to. Returns
``None`` when the URL is acceptable.
"""
raw = (webhook_url or "").strip()
if not raw:
return "Error: webhook_url is empty."
parsed = urlparse(raw)
if parsed.scheme != "https":
return "Error: webhook_url must be an https:// Discord URL."
if "@" in (parsed.netloc or ""):
return "Error: webhook_url must not contain credentials."
host = (parsed.hostname or "").lower().strip(".")
if not (
host in ("discord.com", "discordapp.com")
or host.endswith(".discord.com")
or host.endswith(".discordapp.com")
):
return "Error: webhook_url must be a Discord (discord.com) URL."
return None
TOOL_NAME = "discord_webhooks"
TOOL_DESCRIPTION = (
"Manage and use Discord webhooks. "
"Actions: 'create', 'list', 'delete', 'edit', 'execute'."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": [
"create",
"list",
"delete",
"edit",
"execute",
],
"description": "The webhook action.",
},
"channel_id": {
"type": "string",
"description": ("Channel ID (required for create, list, execute)."),
},
"webhook_id": {
"type": "string",
"description": "Webhook ID (for edit/delete).",
},
"webhook_url": {
"type": "string",
"description": "Full webhook URL (for execute).",
},
"name": {
"type": "string",
"description": "Webhook name (for create/edit).",
},
"content": {
"type": "string",
"description": "Message content (for execute).",
},
"username": {
"type": "string",
"description": ("Override display name (for execute)."),
},
"avatar_url": {
"type": "string",
"description": ("Override avatar URL (for execute)."),
},
"embed_json": {
"type": "string",
"description": ("JSON string of embed data (for execute)."),
},
"server_id": {
"type": "string",
"description": (
"Server ID (for edit/delete when webhook_url " "is not provided)."
),
},
},
"required": ["action"],
}
[docs]
async def run(
action: str,
channel_id: str | None = None,
webhook_id: str | None = None,
webhook_url: str | None = None,
name: str | None = None,
content: str | None = None,
username: str | None = None,
avatar_url: str | None = None,
embed_json: str | None = None,
server_id: str | None = None,
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
action (str): The action value.
channel_id (str | None): Discord/Matrix channel identifier.
webhook_id (str | None): The webhook id value.
webhook_url (str | None): The webhook url value.
name (str | None): Human-readable name.
content (str | None): Content data.
username (str | None): The username value.
avatar_url (str | None): The avatar url value.
embed_json (str | None): The embed json value.
server_id (str | None): The server id value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
import discord
import aiohttp
from tools._discord_helpers import (
require_discord_client,
get_channel,
check_admin_permission,
)
client = require_discord_client(ctx)
if isinstance(client, str):
return client
user_id = ctx.user_id if ctx else ""
# --- CREATE ---
if action == "create":
if not channel_id:
return "Error: channel_id is required for create."
if not name:
return "Error: name is required for create."
channel = await get_channel(client, channel_id)
if isinstance(channel, str):
return channel
if not hasattr(channel, "create_webhook"):
return "Error: This channel doesn't support webhooks."
guild = getattr(channel, "guild", None)
if guild:
ok, err = await check_admin_permission(
client,
user_id,
str(guild.id),
)
if not ok:
return err
try:
wh = await channel.create_webhook(name=name)
return f"Created webhook '{wh.name}' (ID: {wh.id})\n" f"URL: {wh.url}"
except discord.errors.Forbidden:
return "Error: No permission to create webhooks."
except Exception as exc:
return f"Error creating webhook: {exc}"
# --- LIST ---
if action == "list":
if not channel_id:
return "Error: channel_id is required for list."
channel = await get_channel(client, channel_id)
if isinstance(channel, str):
return channel
if not hasattr(channel, "webhooks"):
return "Error: This channel doesn't support webhooks."
try:
webhooks = await channel.webhooks()
except discord.errors.Forbidden:
return "Error: No permission to list webhooks."
except Exception as exc:
return f"Error listing webhooks: {exc}"
if not webhooks:
ch_name = getattr(channel, "name", channel_id)
return f"No webhooks found in '{ch_name}'."
lines = [f"Webhooks in " f"'{getattr(channel, 'name', channel_id)}':"]
for wh in webhooks:
lines.append(f" - {wh.name} (ID: {wh.id}); webhook URL omitted (secret)")
return "\n".join(lines)
# --- DELETE ---
if action == "delete":
if not webhook_id and not webhook_url:
return "Error: webhook_id or webhook_url is required " "for delete."
if webhook_url:
url_err = _validate_discord_webhook_url(webhook_url)
if url_err:
return url_err
try:
async with aiohttp.ClientSession() as s:
wh = discord.Webhook.from_url(
webhook_url,
session=s,
)
await wh.delete()
return "Webhook deleted successfully."
except Exception as exc:
return f"Error deleting webhook by URL: {exc}"
# Delete by ID - need to find it in a server
sid = server_id or (ctx.guild_id if ctx else "")
if not sid:
return "Error: server_id is required when deleting " "by webhook_id."
ok, err = await check_admin_permission(
client,
user_id,
sid,
)
if not ok:
return err
try:
guild = client.get_guild(int(sid))
except ValueError:
return f"Error: Invalid server ID: '{sid}'."
if not guild:
return f"Error: Server '{sid}' not found."
try:
webhooks = await guild.webhooks()
wh = next(
(w for w in webhooks if str(w.id) == webhook_id),
None,
)
if not wh:
return f"Error: Webhook '{webhook_id}' not found."
wname = wh.name
await wh.delete()
return f"Deleted webhook '{wname}'."
except discord.errors.Forbidden:
return "Error: No permission to delete webhooks."
except Exception as exc:
return f"Error deleting webhook: {exc}"
# --- EDIT ---
if action == "edit":
if not webhook_id:
return "Error: webhook_id is required for edit."
if not name:
return "Error: name is required for edit."
sid = server_id or (ctx.guild_id if ctx else "")
if not sid:
return "Error: server_id is required for edit."
ok, err = await check_admin_permission(
client,
user_id,
sid,
)
if not ok:
return err
try:
guild = client.get_guild(int(sid))
except ValueError:
return f"Error: Invalid server ID: '{sid}'."
if not guild:
return f"Error: Server '{sid}' not found."
try:
webhooks = await guild.webhooks()
wh = next(
(w for w in webhooks if str(w.id) == webhook_id),
None,
)
if not wh:
return f"Error: Webhook '{webhook_id}' not found."
await wh.edit(name=name)
return f"Edited webhook to name '{name}'."
except discord.errors.Forbidden:
return "Error: No permission to edit webhooks."
except Exception as exc:
return f"Error editing webhook: {exc}"
# --- EXECUTE ---
if action == "execute":
if not webhook_url:
return "Error: webhook_url is required for execute."
url_err = _validate_discord_webhook_url(webhook_url)
if url_err:
return url_err
if not content and not embed_json:
return "Error: content or embed_json is required."
embed = None
if embed_json:
try:
data: dict[str, Any] = json.loads(embed_json)
embed = discord.Embed.from_dict(data)
except json.JSONDecodeError as exc:
return f"Error: Invalid embed_json: {exc}"
except Exception as exc:
return f"Error creating embed: {exc}"
try:
async with aiohttp.ClientSession() as s:
wh = discord.Webhook.from_url(
webhook_url,
session=s,
)
kwargs: dict[str, Any] = {"wait": True}
if content:
kwargs["content"] = content
if username:
kwargs["username"] = username
if avatar_url:
kwargs["avatar_url"] = avatar_url
if embed:
kwargs["embed"] = embed
msg = await wh.send(**kwargs)
return f"Webhook message sent (ID: {msg.id})."
except Exception as exc:
return f"Error executing webhook: {exc}"
return f"Error: Unknown action '{action}'."