"""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 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:
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:
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:
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:
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:
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,
},
]