"""Discord user API tools using per-user OAuth tokens.
Provides access to the user's own Discord profile, guilds, and connected
accounts using their personal OAuth token (not the bot's token). Requires
the user to have connected their Discord account via the OAuth flow.
"""
from __future__ import annotations
import jsonutil as json
import logging
from typing import Any, TYPE_CHECKING
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
DISCORD_API = "https://discord.com/api/v10"
async def _discord_request(
method: str,
path: str,
token: str,
*,
params: dict[str, Any] | None = None,
) -> dict[str, Any] | list[Any] | str:
"""Issue a single authenticated request against the Discord REST API.
Low-level HTTP helper shared by every handler in this module. Opens a
fresh ``aiohttp.ClientSession``, sends *method* to ``DISCORD_API + path``
with the caller's bearer *token* and optional query *params*, and decodes
the response body. HTTP status codes of 400 and above are not raised; they
are folded into a structured error dict (with a truncated detail snippet)
so callers can surface the failure as JSON rather than crashing.
This function performs outbound HTTP I/O to ``discord.com`` but mutates no
local state, Redis, or knowledge-graph data. The bearer token is the
user's personal OAuth token (resolved upstream by :func:`_get_token`), not
the bot token. It is called by :func:`discord_user_profile`,
:func:`discord_user_guilds`, and :func:`discord_user_connections`; no
other modules invoke it.
Args:
method: HTTP verb to use (e.g. ``"GET"``).
path: API path appended to :data:`DISCORD_API` (e.g.
``"/users/@me"``); it should begin with a leading slash.
token: The user's Discord OAuth bearer token, sent in the
``Authorization`` header.
params: Optional query-string parameters passed to the request.
Returns:
dict | list | str: The parsed JSON body on success (a ``dict`` or
``list`` depending on the endpoint). On an HTTP error status, a
``dict`` with ``error`` and a truncated ``detail`` string. If the body
is not valid JSON, the raw text truncated to 4000 characters.
"""
headers = {"Authorization": f"Bearer {token}"}
url = f"{DISCORD_API}{path}"
async with aiohttp.ClientSession() as session:
async with session.request(method, url, headers=headers, params=params) as resp:
body = await resp.text()
if resp.status >= 400:
return {
"error": f"Discord API error ({resp.status})",
"detail": body[:2000],
}
try:
return json.loads(body)
except json.JSONDecodeError:
return body[:4000]
async def _get_token(ctx: ToolContext | None) -> str:
"""Resolve the calling user's stored Discord OAuth token.
Thin wrapper that validates the tool context, then delegates to
``oauth_manager.require_oauth_token`` to fetch (and refresh, if expired)
the user's personal Discord access token. The OAuth manager reads the
token material from Redis via ``ctx.redis``, so a context without a Redis
client cannot succeed.
The import of ``oauth_manager`` is deferred to call time to avoid a
module-level import cycle. This helper is private to this module and is
called by :func:`discord_user_profile`, :func:`discord_user_guilds`, and
:func:`discord_user_connections`; same-named helpers exist independently in
other OAuth tool modules (``microsoft_tools``, ``google_oauth_tools``,
``github_tools``) and are unrelated to this one.
Args:
ctx: The tool ``ToolContext``; must be non-``None`` and carry a usable
``redis`` client for the lookup to proceed.
Returns:
str: The user's current Discord OAuth bearer token.
Raises:
RuntimeError: If *ctx* is ``None`` or has no Redis client.
oauth_manager.OAuthNotConnected: Propagated from
``require_oauth_token`` when the user has not connected their
Discord account (callers catch this to return a friendly message).
"""
if ctx is None or ctx.redis is None:
raise RuntimeError("Context or Redis not available")
from oauth_manager import require_oauth_token
return await require_oauth_token(ctx, "discord")
[docs]
async def discord_user_profile(ctx: ToolContext | None = None) -> str:
"""Fetch the user's own Discord profile via their OAuth token.
Args:
ctx: Tool execution context, used to resolve the user's
personal Discord OAuth token.
Returns:
JSON string with profile fields (id, username, email, avatar,
Nitro/premium type, etc.), or an error/not-connected message.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
data = await _discord_request("GET", "/users/@me", token)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
return json.dumps(
{
"id": data.get("id"),
"username": data.get("username"),
"global_name": data.get("global_name"),
"discriminator": data.get("discriminator"),
"email": data.get("email"),
"verified": data.get("verified"),
"avatar": data.get("avatar"),
"banner": data.get("banner"),
"accent_color": data.get("accent_color"),
"premium_type": data.get("premium_type"),
"flags": data.get("public_flags"),
}
)
[docs]
async def discord_user_guilds(
limit: int = 50,
ctx: ToolContext | None = None,
) -> str:
"""List Discord guilds the user belongs to via their OAuth token.
Args:
limit: Maximum number of guilds to return (capped at 200).
ctx: Tool execution context, used to resolve the user's
personal Discord OAuth token.
Returns:
JSON string with a guild count and list (id, name, icon,
owner, permissions, features), or an error/not-connected
message.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
params = {"limit": min(limit, 200)}
data = await _discord_request("GET", "/users/@me/guilds", token, params=params)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
guilds = []
for g in data if isinstance(data, list) else []:
guilds.append(
{
"id": g.get("id"),
"name": g.get("name"),
"icon": g.get("icon"),
"owner": g.get("owner"),
"permissions": g.get("permissions"),
"features": g.get("features", []),
}
)
return json.dumps({"count": len(guilds), "guilds": guilds})
[docs]
async def discord_user_connections(ctx: ToolContext | None = None) -> str:
"""List the user's connected accounts via their OAuth token.
Args:
ctx: Tool execution context, used to resolve the user's
personal Discord OAuth token.
Returns:
JSON string with a count and list of connected accounts
(type, name, id, verified, visibility), or an
error/not-connected message.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
data = await _discord_request("GET", "/users/@me/connections", token)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
connections = []
for c in data if isinstance(data, list) else []:
connections.append(
{
"type": c.get("type"),
"name": c.get("name"),
"id": c.get("id"),
"verified": c.get("verified"),
"visibility": c.get("visibility"),
}
)
return json.dumps({"count": len(connections), "connections": connections})
# ---------------------------------------------------------------------------
# TOOLS registration
# ---------------------------------------------------------------------------
TOOLS = [
{
"name": "discord_user_profile",
"description": "Get the user's own Discord profile including username, email, avatar, and Nitro status. Uses the user's personal OAuth token.",
"parameters": {
"type": "object",
"properties": {},
},
"handler": discord_user_profile,
},
{
"name": "discord_user_guilds",
"description": "List Discord servers (guilds) the user is a member of, using their personal OAuth token.",
"parameters": {
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Max guilds to return (default 50, max 200)",
},
},
},
"handler": discord_user_guilds,
},
{
"name": "discord_user_connections",
"description": "List the user's connected accounts on Discord (Twitch, Steam, Spotify, YouTube, etc.). Uses their personal OAuth token.",
"parameters": {
"type": "object",
"properties": {},
},
"handler": discord_user_connections,
},
]