diff --git a/gateway/channel_directory.py b/gateway/channel_directory.py
index b5bab5199e750..e3c19d8eb1541 100644
--- a/gateway/channel_directory.py
+++ b/gateway/channel_directory.py
@@ -151,6 +151,12 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
 
     for platform, adapter in adapters.items():
         try:
+            list_channels = getattr(adapter, "list_channels", None)
+            if callable(list_channels):
+                platform_channels = await list_channels()
+                if platform_channels is not None:
+                    platforms[platform.value] = _normalize_adapter_channels(platform_channels)
+                    continue
             if platform == Platform.DISCORD:
                 platforms["discord"] = await asyncio.to_thread(_build_discord, adapter)
             elif platform == Platform.SLACK:
@@ -259,6 +265,34 @@ def _slack_api_error_code(error: Exception) -> Optional[str]:
     return None
 
 
+def _normalize_adapter_channels(raw_channels: Any) -> List[Dict[str, Any]]:
+    """Validate and dedupe channel entries returned by an adapter's
+    ``list_channels()`` hook (see ``build_channel_directory``)."""
+    channels: List[Dict[str, Any]] = []
+    seen_ids = set()
+    if not isinstance(raw_channels, list):
+        return channels
+    for raw in raw_channels:
+        if not isinstance(raw, dict):
+            continue
+        channel_id = str(raw.get("id") or "").strip()
+        name = str(raw.get("name") or channel_id).strip()
+        if not channel_id or not name or channel_id in seen_ids:
+            continue
+        entry: Dict[str, Any] = {
+            "id": channel_id,
+            "name": name,
+            "type": str(raw.get("type") or "dm"),
+        }
+        if raw.get("thread_id"):
+            entry["thread_id"] = str(raw.get("thread_id"))
+        if raw.get("guild"):
+            entry["guild"] = str(raw.get("guild"))
+        channels.append(entry)
+        seen_ids.add(channel_id)
+    return channels
+
+
 async def _build_slack(adapter) -> List[Dict[str, Any]]:
     """List Slack channels the bot has joined across all workspaces.
 
diff --git a/tests/gateway/test_channel_directory.py b/tests/gateway/test_channel_directory.py
index 634bacc75e091..8df58f36914a9 100644
--- a/tests/gateway/test_channel_directory.py
+++ b/tests/gateway/test_channel_directory.py
@@ -7,6 +7,7 @@ import threading
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
+from gateway.config import Platform
 from gateway.channel_directory import (
     build_channel_directory,
     lookup_channel_type,
@@ -70,6 +71,25 @@ class TestBuildChannelDirectoryWrites:
 
         assert result == previous
 
+    def test_uses_adapter_list_channels_when_available(self, tmp_path):
+        class AdapterWithChannels:
+            async def list_channels(self):
+                return [
+                    {"id": "default", "name": "主对话", "type": "dm"},
+                    {"id": "family_1", "name": "达拉崩吧", "type": "group"},
+                    {"id": "", "name": "ignored", "type": "dm"},
+                    {"id": "family_1", "name": "duplicate", "type": "group"},
+                ]
+
+        cache_file = tmp_path / "channel_directory.json"
+        with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file):
+            directory = asyncio.run(build_channel_directory({Platform.TELEGRAM: AdapterWithChannels()}))
+
+        assert directory["platforms"]["telegram"] == [
+            {"id": "default", "name": "主对话", "type": "dm"},
+            {"id": "family_1", "name": "达拉崩吧", "type": "group"},
+        ]
+
 
 class TestBuildChannelDirectoryOffload:
     def test_discord_builder_runs_off_event_loop_thread(self, tmp_path):