"""Scan the bot's guild cache and return a comprehensive index of all connected servers."""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "force_guild_index"
TOOL_DESCRIPTION = (
"Scan the bot's internal cache for all connected guilds and return an index "
"with ID, name, member count, and owner ID. Also caches to Redis."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {},
}
[docs]
async def run(ctx: ToolContext | None = None) -> str:
"""Execute this tool and return the result.
Args:
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
from tools._discord_helpers import require_discord_client
client = require_discord_client(ctx)
if isinstance(client, str):
return client
redis_client = getattr(ctx, "redis", None)
guild_index = []
for guild in client.guilds:
guild_info = {
"id": str(guild.id),
"name": guild.name,
"member_count": guild.member_count,
"owner_id": str(guild.owner_id) if guild.owner_id else "Unknown",
}
guild_index.append(guild_info)
if redis_client:
pipe = redis_client.pipeline(transaction=True)
for info in guild_index:
pipe.set(
f"guild_registry:{info['id']}",
str({
"id": info["id"],
"name": info["name"],
"member_count": info["member_count"],
"owner_id": info["owner_id"],
}),
)
pipe.set("guild_registry:full", json.dumps(guild_index))
await pipe.execute()
guild_index.sort(key=lambda x: x["member_count"] or 0, reverse=True)
return json.dumps(guild_index, indent=2)