"""List all servers/guilds and channels the bot is active in.
╔═══════════════════════════════════════════════════════════════════════════════╗
║ 🌐 CROSS-PLATFORM SERVER LISTING ║
╠═══════════════════════════════════════════════════════════════════════════════╣
║ list_active_servers — Shows every server, guild, and channel across all ║
║ connected platforms (Discord, Matrix, …) ║
╚═══════════════════════════════════════════════════════════════════════════════╝
"""
from __future__ import annotations
import jsonutil as json
import logging
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "list_active_servers"
TOOL_DESCRIPTION = (
"List ALL servers, guilds, and channels the bot is currently active "
"in across every connected platform (Discord, Matrix, etc.). "
"Returns a comprehensive inventory grouped by platform, with each "
"server's channels listed under it."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {},
}
[docs]
async def run(ctx: "ToolContext | None" = None) -> str:
"""Return a JSON inventory of every platform, server, and channel.
Entry point for the ``list_active_servers`` tool. It gives the bot a
cross-platform view of where it is present -- every Discord guild, Matrix
room, and so on -- grouped by platform with each server's channels nested
underneath, so the model can reason about where it can act.
Prefers a fast Redis path: it scans ``sg:platform:inventory:*`` keys on
``ctx.redis`` and reconstructs the inventory from those cached blobs,
deriving the platform name from the key suffix. If the cache is empty or
unavailable, it falls back to live adapters from ``ctx.all_adapters``,
awaiting each running adapter's ``list_servers_and_channels()`` and
recording per-platform errors rather than failing the whole call. Read-only
apart from those Redis reads; per-platform exceptions are logged via the
module ``logger``. Dispatched through the tool loader's
``getattr(module, "run")`` path (see ``tool_loader.py``) and exercised
directly by ``tests/core/migration/test_list_active_servers_caching.py``.
Args:
ctx: The :class:`ToolContext`; ``redis`` and ``all_adapters`` are read
from it. ``None`` yields an error envelope.
Returns:
str: A pretty-printed JSON object with ``success``, ``platform_count``
and a ``platforms`` list (each entry carrying ``running``, its
``servers``, and ``server_count`` / ``total_channels`` or an
``error``); or an error envelope when no context or no adapters are
available.
"""
if ctx is None:
return json.dumps({"success": False, "error": "No tool context."})
# Try querying from Redis platform inventory cache first
if ctx.redis is not None:
try:
keys = await ctx.redis.keys("sg:platform:inventory:*")
if keys:
platforms_data = []
for key in keys:
key_str = key.decode() if isinstance(key, bytes) else str(key)
platform_name = key_str.split(":")[-1]
raw_data = await ctx.redis.get(key)
if raw_data:
servers = json.loads(raw_data.decode() if isinstance(raw_data, bytes) else raw_data)
platforms_data.append({
"platform": platform_name,
"running": True,
"servers": servers,
"server_count": len(servers),
"total_channels": sum(len(s.get("channels", [])) for s in servers)
})
if platforms_data:
return json.dumps({
"success": True,
"platform_count": len(platforms_data),
"platforms": platforms_data
}, indent=2)
except Exception as exc:
logger.exception("Failed to query platform inventory from Redis cache")
adapters = getattr(ctx, "all_adapters", None) or []
if not adapters:
return json.dumps(
{
"success": False,
"error": "No platform adapters available.",
}
)
result: dict[str, Any] = {
"success": True,
"platform_count": 0,
"platforms": [],
}
for adapter in adapters:
platform_name = getattr(adapter, "name", "unknown")
is_running = getattr(adapter, "is_running", False)
platform_entry: dict[str, Any] = {
"platform": platform_name,
"running": is_running,
"servers": [],
}
if is_running:
try:
servers = await adapter.list_servers_and_channels()
platform_entry["servers"] = servers
platform_entry["server_count"] = len(servers)
platform_entry["total_channels"] = sum(
len(s.get("channels", [])) for s in servers
)
except Exception as exc:
logger.exception(
"Failed to list servers for platform %s",
platform_name,
)
platform_entry["error"] = str(exc)
result["platforms"].append(platform_entry)
result["platform_count"] = len(result["platforms"])
return json.dumps(result, indent=2)