Discord

{
  "name": "discord",
  "version": "1.0.0",
  "description": "Build and automate Discord integrations (bots, webhooks, slash commands, and REST API workflows). Use when the user mentions Discord, a Discord server/guild, channels, webhooks, bot tokens, slash c...",
  "author": {
    "name": "OpenHands",
    "email": "contact@all-hands.dev"
  },
  "homepage": "https://github.com/OpenHands/extensions",
  "repository": "https://github.com/OpenHands/extensions",
  "license": "MIT",
  "keywords": [
    "discord",
    "bot",
    "webhook",
    "automation"
  ]
}

from __future__ import annotations

import random
import time
from dataclasses import dataclass
from typing import Any, Mapping

import requests


@dataclass(frozen=True)
class DiscordRateLimitInfo:
    retry_after: float
    is_global: bool
    bucket: str | None
    remaining: str | None
    reset_after: str | None


class DiscordHTTPError(RuntimeError):
    pass


def _parse_rate_limit_info(*, status_code: int, headers: Mapping[str, str], json_body: Any) -> DiscordRateLimitInfo | None:
    if status_code != 429:
        return None

    retry_after: float | None = None
    is_global = False

    if isinstance(json_body, dict):
        retry_after_val = json_body.get("retry_after")
        if isinstance(retry_after_val, (int, float, str)):
            try:
                retry_after = float(retry_after_val)
            except ValueError:
                retry_after = None
        is_global = bool(json_body.get("global", False))

    if retry_after is None:
        hdr = headers.get("Retry-After")
        if hdr is not None:
            try:
                retry_after = float(hdr)
            except ValueError:
                retry_after = None

    if retry_after is None:
        reset_after = headers.get("X-RateLimit-Reset-After")
        if reset_after is not None:
            try:
                retry_after = float(reset_after)
            except ValueError:
                retry_after = None

    if retry_after is None:
        return None

    return DiscordRateLimitInfo(
        retry_after=retry_after,
        is_global=is_global,
        bucket=headers.get("X-RateLimit-Bucket"),
        remaining=headers.get("X-RateLimit-Remaining"),
        reset_after=headers.get("X-RateLimit-Reset-After"),
    )


def post_json(
    *,
    url: str,
    headers: Mapping[str, str],
    payload: Mapping[str, object],
    timeout_s: float = 30,
    max_retries: int = 3,
    max_retry_after_s: float = 60.0,
    jitter_s: float = 0.25,
    redact_url_in_errors: bool = False,
) -> dict[str, object] | None:
    attempt = 0
    while True:
        attempt += 1
        try:
            resp = requests.post(url, headers=dict(headers), json=dict(payload), timeout=timeout_s)
        except requests.RequestException as e:
            raise DiscordHTTPError(f"HTTP request failed ({e}).") from e

        body_text = resp.text or ""
        json_body: Any = None
        if body_text:
            try:
                json_body = resp.json()
            except ValueError:
                json_body = None

        rl = _parse_rate_limit_info(status_code=resp.status_code, headers=resp.headers, json_body=json_body)
        if rl is not None and attempt <= max_retries:
            sleep_s = min(max(0.0, rl.retry_after), max_retry_after_s)
            if jitter_s > 0:
                sleep_s += random.uniform(0.0, jitter_s)
            time.sleep(sleep_s)
            continue

        if resp.status_code >= 400:
            context_bits: list[str] = []
            if not redact_url_in_errors:
                context_bits.append(f"url={url}")
            if rl is not None:
                context_bits.append(f"rate_limit_global={rl.is_global}")
                if rl.bucket is not None:
                    context_bits.append(f"rate_limit_bucket={rl.bucket}")
                if rl.remaining is not None:
                    context_bits.append(f"rate_limit_remaining={rl.remaining}")
                if rl.reset_after is not None:
                    context_bits.append(f"rate_limit_reset_after={rl.reset_after}")

            context = (" " + " ".join(context_bits)) if context_bits else ""

            msg = f"HTTP request failed (HTTP {resp.status_code}).{context}"
            if body_text:
                msg += f" Response: {body_text[:500]}"
            raise DiscordHTTPError(msg)

        if not body_text:
            return None
        if isinstance(json_body, dict):
            return json_body
        return {"raw": body_text}

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.parse
import urllib.request

from ._http import DiscordHTTPError, post_json


def _with_wait_param(url: str, *, wait: bool) -> str:
    if not wait:
        return url

    parts = urllib.parse.urlsplit(url)
    query = dict(urllib.parse.parse_qsl(parts.query, keep_blank_values=True))
    query["wait"] = "true"

    return urllib.parse.urlunsplit(
        (parts.scheme, parts.netloc, parts.path, urllib.parse.urlencode(query), parts.fragment)
    )


def _request_json(url: str, payload: dict[str, object], *, wait: bool, max_retries: int) -> dict[str, object] | None:
    request_url = _with_wait_param(url, wait=wait)

    headers = {
        "Content-Type": "application/json",
        "User-Agent": "OpenHands-DiscordSkill/1.0 (+https://github.com/OpenHands/skills)",
    }

    return post_json(
        url=request_url,
        headers=headers,
        payload=payload,
        max_retries=max_retries,
        redact_url_in_errors=True,
    )


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Post a message to a Discord incoming webhook. "
            "The webhook URL is secret; avoid printing/logging it."
        )
    )
    parser.add_argument(
        "--webhook-url",
        default=os.getenv("DISCORD_WEBHOOK_URL"),
        help="Incoming webhook URL (default: $DISCORD_WEBHOOK_URL)",
    )
    parser.add_argument(
        "--content",
        help="Message content (max 2000 characters). If omitted, read from stdin.",
    )
    parser.add_argument(
        "--wait",
        action="store_true",
        help="Add ?wait=true to get the created message object.",
    )
    parser.add_argument(
        "--max-retries",
        type=int,
        default=3,
        help="Retries on HTTP 429 (default: 3).",
    )

    args = parser.parse_args()

    if not args.webhook_url:
        print("Missing --webhook-url (or set DISCORD_WEBHOOK_URL).", file=sys.stderr)
        return 2

    content = args.content
    if content is None:
        content = sys.stdin.read().strip()

    if not content:
        print("No content provided (use --content or stdin).", file=sys.stderr)
        return 2

    payload = {
        "content": content,
        "allowed_mentions": {"parse": []},
    }

    result = _request_json(
        args.webhook_url,
        payload,
        wait=args.wait,
        max_retries=max(0, args.max_retries),
    )

    if result is not None:
        print(json.dumps(result, indent=2, sort_keys=True))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.request

from ._http import DiscordHTTPError, post_json


API_BASE = "https://discord.com/api/v10"


def _post_message(
    *,
    token: str,
    channel_id: str,
    payload: dict[str, object],
    max_retries: int,
) -> dict[str, object] | None:
    url = f"{API_BASE}/channels/{channel_id}/messages"

    headers = {
        "Authorization": f"Bot {token}",
        "Content-Type": "application/json",
        "User-Agent": "OpenHands-DiscordSkill/1.0 (+https://github.com/OpenHands/skills)",
    }

    try:
        return post_json(url=url, headers=headers, payload=payload, max_retries=max_retries)
    except DiscordHTTPError as e:
        raise DiscordHTTPError(f"Discord API call failed. channel_id={channel_id}. {e}") from e


def main() -> int:
    parser = argparse.ArgumentParser(description="Send a message to a Discord channel using a bot token.")
    parser.add_argument(
        "--token",
        default=os.getenv("DISCORD_BOT_TOKEN"),
        help="Bot token (default: $DISCORD_BOT_TOKEN)",
    )
    parser.add_argument(
        "--channel-id",
        default=os.getenv("DISCORD_CHANNEL_ID"),
        help="Channel ID (default: $DISCORD_CHANNEL_ID)",
    )
    parser.add_argument(
        "--content",
        help="Message content (max 2000 characters). If omitted, read from stdin.",
    )
    parser.add_argument(
        "--max-retries",
        type=int,
        default=3,
        help="Retries on HTTP 429 (default: 3).",
    )
    parser.add_argument(
        "--allow-mentions",
        action="store_true",
        help="Allow Discord to parse mentions. Default is safe (no mentions).",
    )

    args = parser.parse_args()

    if not args.token:
        print("Missing --token (or set DISCORD_BOT_TOKEN).", file=sys.stderr)
        return 2

    if not args.channel_id:
        print("Missing --channel-id (or set DISCORD_CHANNEL_ID).", file=sys.stderr)
        return 2

    content = args.content
    if content is None:
        content = sys.stdin.read().strip()

    if not content:
        print("No content provided (use --content or stdin).", file=sys.stderr)
        return 2

    payload: dict[str, object] = {"content": content}
    if not args.allow_mentions:
        payload["allowed_mentions"] = {"parse": []}

    result = _post_message(
        token=args.token,
        channel_id=args.channel_id,
        payload=payload,
        max_retries=max(0, args.max_retries),
    )

    if result is not None:
        print(json.dumps(result, indent=2, sort_keys=True))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())