"""Discord webhook management and execution."""
from __future__ import annotations
import json
import logging
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
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:
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."
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}'."