feat: add Discord server introspection and management tool (#4753)

* feat: add Discord server introspection and management tool

Add a discord_server tool that gives the agent the ability to interact
with Discord servers when running on the Discord gateway. Uses Discord
REST API directly with the bot token — no dependency on the gateway
adapter's discord.py client.

The tool is only included in the hermes-discord toolset (zero cost for
users on other platforms) and gated on DISCORD_BOT_TOKEN via check_fn.

Actions (14):
- Introspection: list_guilds, server_info, list_channels, channel_info,
  list_roles, member_info, search_members
- Messages: fetch_messages, list_pins, pin_message, unpin_message
- Management: create_thread, add_role, remove_role

This addresses a gap where users on Discord could not ask Hermes to
review server structure, channels, roles, or members — a task competing
agents (OpenClaw) handle out of the box.

Files changed:
- tools/discord_tool.py (new): Tool implementation + registration
- model_tools.py: Add to discovery list
- toolsets.py: Add to hermes-discord toolset only
- tests/tools/test_discord_tool.py (new): 43 tests covering all actions,
  validation, error handling, registration, and toolset scoping

* feat(discord): intent-aware schema filtering + config allowlist + schema cleanup

- _detect_capabilities() hits GET /applications/@me once per process
  to read GUILD_MEMBERS / MESSAGE_CONTENT privileged intent bits.
- Schema is rebuilt per-session in model_tools.get_tool_definitions:
  hides search_members / member_info when GUILD_MEMBERS intent is off,
  annotates fetch_messages description when MESSAGE_CONTENT is off.
- New config key discord.server_actions (comma-separated or YAML list)
  lets users restrict which actions the agent can call, intersected
  with intent availability. Unknown names are warned and dropped.
- Defense-in-depth: runtime handler re-checks the allowlist so a stale
  cached schema cannot bypass a tightened config.
- Schema description rewritten as an action-first manifest (signature
  per action) instead of per-parameter 'required for X, Y, Z' cross-refs.
  ~25% shorter; model can see each action's required params at a glance.
- Added bounds: limit gets minimum=1 maximum=100, auto_archive_duration
  becomes an enum of the 4 valid Discord values.
- 403 enrichment: runtime 403 errors are mapped to actionable guidance
  (which permission is missing and what to do about it) instead of the
  raw Discord error body.
- 36 new tests: capability detection with caching and force refresh,
  config allowlist parsing (string/list/invalid/unknown), intent+allowlist
  intersection, dynamic schema build, runtime allowlist enforcement,
  403 enrichment, and model_tools integration wiring.
This commit is contained in:
Teknium 2026-04-19 11:52:19 -07:00 committed by GitHub
parent d48d6fadff
commit ef73367fc5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 1907 additions and 1 deletions

View file

@ -712,6 +712,14 @@ DEFAULT_CONFIG = {
"auto_thread": True, # Auto-create threads on @mention in channels (like Slack) "auto_thread": True, # Auto-create threads on @mention in channels (like Slack)
"reactions": True, # Add 👀/✅/❌ reactions to messages during processing "reactions": True, # Add 👀/✅/❌ reactions to messages during processing
"channel_prompts": {}, # Per-channel ephemeral system prompts (forum parents apply to child threads) "channel_prompts": {}, # Per-channel ephemeral system prompts (forum parents apply to child threads)
# discord_server tool: restrict which actions the agent may call.
# Default (empty) = all actions allowed (subject to bot privileged intents).
# Accepts comma-separated string ("list_guilds,list_channels,fetch_messages")
# or YAML list. Unknown names are dropped with a warning at load time.
# Actions: list_guilds, server_info, list_channels, channel_info,
# list_roles, member_info, search_members, fetch_messages, list_pins,
# pin_message, unpin_message, create_thread, add_role, remove_role.
"server_actions": "",
}, },
# WhatsApp platform settings (gateway mode) # WhatsApp platform settings (gateway mode)

View file

@ -282,6 +282,31 @@ def get_tool_definitions(
filtered_tools[i] = {"type": "function", "function": dynamic_schema} filtered_tools[i] = {"type": "function", "function": dynamic_schema}
break break
# Rebuild discord_server schema based on the bot's privileged intents
# (detected from GET /applications/@me) and the user's action allowlist
# in config. Hides actions the bot's intents don't support so the
# model never attempts them, and annotates fetch_messages when the
# MESSAGE_CONTENT intent is missing.
if "discord_server" in available_tool_names:
try:
from tools.discord_tool import get_dynamic_schema
dynamic = get_dynamic_schema()
except Exception: # pragma: no cover — defensive, fall back to static
dynamic = None
if dynamic is None:
# Tool filtered out entirely (empty allowlist or detection disabled
# the only remaining actions). Drop it from the schema list.
filtered_tools = [
t for t in filtered_tools
if t.get("function", {}).get("name") != "discord_server"
]
available_tool_names.discard("discord_server")
else:
for i, td in enumerate(filtered_tools):
if td.get("function", {}).get("name") == "discord_server":
filtered_tools[i] = {"type": "function", "function": dynamic}
break
# Strip web tool cross-references from browser_navigate description when # Strip web tool cross-references from browser_navigate description when
# web_search / web_extract are not available. The static schema says # web_search / web_extract are not available. The static schema says
# "prefer web_search or web_extract" which causes the model to hallucinate # "prefer web_search or web_extract" which causes the model to hallucinate

View file

@ -0,0 +1,979 @@
"""Tests for the Discord server introspection and management tool."""
import json
import os
import urllib.error
from io import BytesIO
from unittest.mock import MagicMock, patch
import pytest
from tools.discord_tool import (
DiscordAPIError,
_ACTIONS,
_available_actions,
_build_schema,
_channel_type_name,
_detect_capabilities,
_discord_request,
_enrich_403,
_get_bot_token,
_load_allowed_actions_config,
_reset_capability_cache,
check_discord_tool_requirements,
discord_server,
get_dynamic_schema,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _mock_urlopen(response_data, status=200):
"""Create a mock for urllib.request.urlopen."""
mock_resp = MagicMock()
mock_resp.status = status
mock_resp.read.return_value = json.dumps(response_data).encode("utf-8")
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
mock_resp.__exit__ = MagicMock(return_value=False)
return mock_resp
# ---------------------------------------------------------------------------
# Token / check_fn
# ---------------------------------------------------------------------------
class TestCheckRequirements:
def test_no_token(self, monkeypatch):
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
assert check_discord_tool_requirements() is False
def test_empty_token(self, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "")
assert check_discord_tool_requirements() is False
def test_valid_token(self, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token-123")
assert check_discord_tool_requirements() is True
def test_get_bot_token(self, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", " my-token ")
assert _get_bot_token() == "my-token"
def test_get_bot_token_missing(self, monkeypatch):
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
assert _get_bot_token() is None
# ---------------------------------------------------------------------------
# Channel type names
# ---------------------------------------------------------------------------
class TestChannelTypeNames:
def test_known_types(self):
assert _channel_type_name(0) == "text"
assert _channel_type_name(2) == "voice"
assert _channel_type_name(4) == "category"
assert _channel_type_name(5) == "announcement"
assert _channel_type_name(13) == "stage"
assert _channel_type_name(15) == "forum"
def test_unknown_type(self):
assert _channel_type_name(99) == "unknown(99)"
# ---------------------------------------------------------------------------
# Discord API request helper
# ---------------------------------------------------------------------------
class TestDiscordRequest:
@patch("tools.discord_tool.urllib.request.urlopen")
def test_get_request(self, mock_urlopen_fn):
mock_urlopen_fn.return_value = _mock_urlopen({"ok": True})
result = _discord_request("GET", "/test", "token123")
assert result == {"ok": True}
# Verify the request was constructed correctly
call_args = mock_urlopen_fn.call_args
req = call_args[0][0]
assert "https://discord.com/api/v10/test" in req.full_url
assert req.get_header("Authorization") == "Bot token123"
assert req.get_method() == "GET"
@patch("tools.discord_tool.urllib.request.urlopen")
def test_get_with_params(self, mock_urlopen_fn):
mock_urlopen_fn.return_value = _mock_urlopen({"ok": True})
_discord_request("GET", "/test", "tok", params={"foo": "bar"})
req = mock_urlopen_fn.call_args[0][0]
assert "foo=bar" in req.full_url
@patch("tools.discord_tool.urllib.request.urlopen")
def test_post_with_body(self, mock_urlopen_fn):
mock_urlopen_fn.return_value = _mock_urlopen({"id": "123"})
result = _discord_request("POST", "/channels", "tok", body={"name": "test"})
assert result == {"id": "123"}
req = mock_urlopen_fn.call_args[0][0]
assert req.data == json.dumps({"name": "test"}).encode("utf-8")
@patch("tools.discord_tool.urllib.request.urlopen")
def test_204_returns_none(self, mock_urlopen_fn):
mock_resp = _mock_urlopen({}, status=204)
mock_urlopen_fn.return_value = mock_resp
result = _discord_request("PUT", "/pins/1", "tok")
assert result is None
@patch("tools.discord_tool.urllib.request.urlopen")
def test_http_error(self, mock_urlopen_fn):
error_body = json.dumps({"message": "Missing Access"}).encode()
http_error = urllib.error.HTTPError(
url="https://discord.com/api/v10/test",
code=403,
msg="Forbidden",
hdrs={},
fp=BytesIO(error_body),
)
mock_urlopen_fn.side_effect = http_error
with pytest.raises(DiscordAPIError) as exc_info:
_discord_request("GET", "/test", "tok")
assert exc_info.value.status == 403
assert "Missing Access" in exc_info.value.body
# ---------------------------------------------------------------------------
# Main handler: validation
# ---------------------------------------------------------------------------
class TestDiscordServerValidation:
def test_no_token(self, monkeypatch):
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
result = json.loads(discord_server(action="list_guilds"))
assert "error" in result
assert "DISCORD_BOT_TOKEN" in result["error"]
def test_unknown_action(self, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
result = json.loads(discord_server(action="bad_action"))
assert "error" in result
assert "Unknown action" in result["error"]
assert "available_actions" in result
def test_missing_required_guild_id(self, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
result = json.loads(discord_server(action="list_channels"))
assert "error" in result
assert "guild_id" in result["error"]
def test_missing_required_channel_id(self, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
result = json.loads(discord_server(action="fetch_messages"))
assert "error" in result
assert "channel_id" in result["error"]
def test_missing_multiple_params(self, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
result = json.loads(discord_server(action="add_role"))
assert "error" in result
assert "guild_id" in result["error"]
assert "user_id" in result["error"]
assert "role_id" in result["error"]
# ---------------------------------------------------------------------------
# Action: list_guilds
# ---------------------------------------------------------------------------
class TestListGuilds:
@patch("tools.discord_tool._discord_request")
def test_list_guilds(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = [
{"id": "111", "name": "Test Server", "icon": "abc", "owner": True, "permissions": "123"},
{"id": "222", "name": "Other Server", "icon": None, "owner": False, "permissions": "456"},
]
result = json.loads(discord_server(action="list_guilds"))
assert result["count"] == 2
assert result["guilds"][0]["name"] == "Test Server"
assert result["guilds"][1]["id"] == "222"
mock_req.assert_called_once_with("GET", "/users/@me/guilds", "test-token")
# ---------------------------------------------------------------------------
# Action: server_info
# ---------------------------------------------------------------------------
class TestServerInfo:
@patch("tools.discord_tool._discord_request")
def test_server_info(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = {
"id": "111",
"name": "My Server",
"description": "A cool server",
"icon": "icon_hash",
"owner_id": "999",
"approximate_member_count": 42,
"approximate_presence_count": 10,
"features": ["COMMUNITY"],
"premium_tier": 2,
"premium_subscription_count": 5,
"verification_level": 1,
}
result = json.loads(discord_server(action="server_info", guild_id="111"))
assert result["name"] == "My Server"
assert result["member_count"] == 42
assert result["online_count"] == 10
mock_req.assert_called_once_with(
"GET", "/guilds/111", "test-token", params={"with_counts": "true"}
)
# ---------------------------------------------------------------------------
# Action: list_channels
# ---------------------------------------------------------------------------
class TestListChannels:
@patch("tools.discord_tool._discord_request")
def test_list_channels_organized(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = [
{"id": "10", "name": "General", "type": 4, "position": 0, "parent_id": None},
{"id": "11", "name": "chat", "type": 0, "position": 0, "parent_id": "10", "topic": "Main chat", "nsfw": False},
{"id": "12", "name": "voice", "type": 2, "position": 1, "parent_id": "10", "topic": None, "nsfw": False},
{"id": "13", "name": "no-category", "type": 0, "position": 0, "parent_id": None, "topic": None, "nsfw": False},
]
result = json.loads(discord_server(action="list_channels", guild_id="111"))
assert result["total_channels"] == 3 # excludes the category itself
groups = result["channel_groups"]
# Uncategorized first
assert groups[0]["category"] is None
assert len(groups[0]["channels"]) == 1
assert groups[0]["channels"][0]["name"] == "no-category"
# Then the category
assert groups[1]["category"]["name"] == "General"
assert len(groups[1]["channels"]) == 2
@patch("tools.discord_tool._discord_request")
def test_empty_guild(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = []
result = json.loads(discord_server(action="list_channels", guild_id="111"))
assert result["total_channels"] == 0
# ---------------------------------------------------------------------------
# Action: channel_info
# ---------------------------------------------------------------------------
class TestChannelInfo:
@patch("tools.discord_tool._discord_request")
def test_channel_info(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = {
"id": "11", "name": "general", "type": 0, "guild_id": "111",
"topic": "Welcome!", "nsfw": False, "position": 0,
"parent_id": "10", "rate_limit_per_user": 0, "last_message_id": "999",
}
result = json.loads(discord_server(action="channel_info", channel_id="11"))
assert result["name"] == "general"
assert result["type"] == "text"
assert result["guild_id"] == "111"
# ---------------------------------------------------------------------------
# Action: list_roles
# ---------------------------------------------------------------------------
class TestListRoles:
@patch("tools.discord_tool._discord_request")
def test_list_roles_sorted(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = [
{"id": "1", "name": "@everyone", "position": 0, "color": 0, "mentionable": False, "managed": False, "hoist": False},
{"id": "2", "name": "Admin", "position": 2, "color": 16711680, "mentionable": True, "managed": False, "hoist": True},
{"id": "3", "name": "Mod", "position": 1, "color": 255, "mentionable": True, "managed": False, "hoist": True},
]
result = json.loads(discord_server(action="list_roles", guild_id="111"))
assert result["count"] == 3
# Should be sorted by position descending
assert result["roles"][0]["name"] == "Admin"
assert result["roles"][0]["color"] == "#ff0000"
assert result["roles"][1]["name"] == "Mod"
assert result["roles"][2]["name"] == "@everyone"
# ---------------------------------------------------------------------------
# Action: member_info
# ---------------------------------------------------------------------------
class TestMemberInfo:
@patch("tools.discord_tool._discord_request")
def test_member_info(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = {
"user": {"id": "42", "username": "testuser", "global_name": "Test User", "avatar": "abc", "bot": False},
"nick": "Testy",
"roles": ["2", "3"],
"joined_at": "2024-01-01T00:00:00Z",
"premium_since": None,
}
result = json.loads(discord_server(action="member_info", guild_id="111", user_id="42"))
assert result["username"] == "testuser"
assert result["nickname"] == "Testy"
assert result["roles"] == ["2", "3"]
# ---------------------------------------------------------------------------
# Action: search_members
# ---------------------------------------------------------------------------
class TestSearchMembers:
@patch("tools.discord_tool._discord_request")
def test_search_members(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = [
{"user": {"id": "42", "username": "testuser", "global_name": "Test", "bot": False}, "nick": None, "roles": []},
]
result = json.loads(discord_server(action="search_members", guild_id="111", query="test"))
assert result["count"] == 1
assert result["members"][0]["username"] == "testuser"
mock_req.assert_called_once_with(
"GET", "/guilds/111/members/search", "test-token",
params={"query": "test", "limit": "50"},
)
@patch("tools.discord_tool._discord_request")
def test_search_members_limit_capped(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = []
discord_server(action="search_members", guild_id="111", query="x", limit=200)
call_params = mock_req.call_args[1]["params"]
assert call_params["limit"] == "100" # Capped at 100
# ---------------------------------------------------------------------------
# Action: fetch_messages
# ---------------------------------------------------------------------------
class TestFetchMessages:
@patch("tools.discord_tool._discord_request")
def test_fetch_messages(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = [
{
"id": "1001",
"content": "Hello world",
"author": {"id": "42", "username": "user1", "global_name": "User One", "bot": False},
"timestamp": "2024-01-01T12:00:00Z",
"edited_timestamp": None,
"attachments": [],
"pinned": False,
},
]
result = json.loads(discord_server(action="fetch_messages", channel_id="11"))
assert result["count"] == 1
assert result["messages"][0]["content"] == "Hello world"
assert result["messages"][0]["author"]["username"] == "user1"
@patch("tools.discord_tool._discord_request")
def test_fetch_messages_with_pagination(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = []
discord_server(action="fetch_messages", channel_id="11", before="999", limit=10)
call_params = mock_req.call_args[1]["params"]
assert call_params["before"] == "999"
assert call_params["limit"] == "10"
# ---------------------------------------------------------------------------
# Action: list_pins
# ---------------------------------------------------------------------------
class TestListPins:
@patch("tools.discord_tool._discord_request")
def test_list_pins(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = [
{"id": "500", "content": "Important announcement", "author": {"username": "admin"}, "timestamp": "2024-01-01T00:00:00Z"},
]
result = json.loads(discord_server(action="list_pins", channel_id="11"))
assert result["count"] == 1
assert result["pinned_messages"][0]["content"] == "Important announcement"
# ---------------------------------------------------------------------------
# Actions: pin_message / unpin_message
# ---------------------------------------------------------------------------
class TestPinUnpin:
@patch("tools.discord_tool._discord_request")
def test_pin_message(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = None # 204
result = json.loads(discord_server(action="pin_message", channel_id="11", message_id="500"))
assert result["success"] is True
mock_req.assert_called_once_with("PUT", "/channels/11/pins/500", "test-token")
@patch("tools.discord_tool._discord_request")
def test_unpin_message(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = None
result = json.loads(discord_server(action="unpin_message", channel_id="11", message_id="500"))
assert result["success"] is True
# ---------------------------------------------------------------------------
# Action: create_thread
# ---------------------------------------------------------------------------
class TestCreateThread:
@patch("tools.discord_tool._discord_request")
def test_create_standalone_thread(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = {"id": "800", "name": "New Thread"}
result = json.loads(discord_server(action="create_thread", channel_id="11", name="New Thread"))
assert result["success"] is True
assert result["thread_id"] == "800"
# Verify the API call
mock_req.assert_called_once_with(
"POST", "/channels/11/threads", "test-token",
body={"name": "New Thread", "auto_archive_duration": 1440, "type": 11},
)
@patch("tools.discord_tool._discord_request")
def test_create_thread_from_message(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = {"id": "801", "name": "Discussion"}
result = json.loads(discord_server(
action="create_thread", channel_id="11", name="Discussion", message_id="1001",
))
assert result["success"] is True
mock_req.assert_called_once_with(
"POST", "/channels/11/messages/1001/threads", "test-token",
body={"name": "Discussion", "auto_archive_duration": 1440},
)
# ---------------------------------------------------------------------------
# Actions: add_role / remove_role
# ---------------------------------------------------------------------------
class TestRoleManagement:
@patch("tools.discord_tool._discord_request")
def test_add_role(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = None
result = json.loads(discord_server(
action="add_role", guild_id="111", user_id="42", role_id="2",
))
assert result["success"] is True
mock_req.assert_called_once_with(
"PUT", "/guilds/111/members/42/roles/2", "test-token",
)
@patch("tools.discord_tool._discord_request")
def test_remove_role(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.return_value = None
result = json.loads(discord_server(
action="remove_role", guild_id="111", user_id="42", role_id="2",
))
assert result["success"] is True
# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------
class TestErrorHandling:
@patch("tools.discord_tool._discord_request")
def test_api_error_handled(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.side_effect = DiscordAPIError(403, '{"message": "Missing Access"}')
result = json.loads(discord_server(action="list_guilds"))
assert "error" in result
assert "403" in result["error"]
@patch("tools.discord_tool._discord_request")
def test_unexpected_error_handled(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
mock_req.side_effect = RuntimeError("something broke")
result = json.loads(discord_server(action="list_guilds"))
assert "error" in result
assert "something broke" in result["error"]
# ---------------------------------------------------------------------------
# Registration
# ---------------------------------------------------------------------------
class TestRegistration:
def test_tool_registered(self):
from tools.registry import registry
entry = registry._tools.get("discord_server")
assert entry is not None
assert entry.schema["name"] == "discord_server"
assert entry.toolset == "discord"
assert entry.check_fn is not None
assert entry.requires_env == ["DISCORD_BOT_TOKEN"]
def test_schema_actions(self):
"""Static schema should list all actions (the model_tools post-processing
narrows this per-session; static registration is the superset)."""
from tools.registry import registry
entry = registry._tools["discord_server"]
actions = entry.schema["parameters"]["properties"]["action"]["enum"]
expected = [
"list_guilds", "server_info", "list_channels", "channel_info",
"list_roles", "member_info", "search_members", "fetch_messages",
"list_pins", "pin_message", "unpin_message", "create_thread",
"add_role", "remove_role",
]
assert set(actions) == set(expected)
assert set(_ACTIONS.keys()) == set(expected)
def test_schema_parameter_bounds(self):
from tools.registry import registry
entry = registry._tools["discord_server"]
props = entry.schema["parameters"]["properties"]
assert props["limit"]["minimum"] == 1
assert props["limit"]["maximum"] == 100
assert props["auto_archive_duration"]["enum"] == [60, 1440, 4320, 10080]
def test_schema_description_is_action_manifest(self):
"""The top-level description should include the action manifest
(one-line signatures per action) so the model can find required
params without re-reading every parameter description."""
from tools.registry import registry
entry = registry._tools["discord_server"]
desc = entry.schema["description"]
# Spot-check a few entries
assert "list_guilds()" in desc
assert "fetch_messages(channel_id)" in desc
assert "add_role(guild_id, user_id, role_id)" in desc
def test_handler_callable(self):
from tools.registry import registry
entry = registry._tools["discord_server"]
assert callable(entry.handler)
# ---------------------------------------------------------------------------
# Toolset: discord_server only in hermes-discord
# ---------------------------------------------------------------------------
class TestToolsetInclusion:
def test_discord_server_in_hermes_discord_toolset(self):
from toolsets import TOOLSETS
assert "discord_server" in TOOLSETS["hermes-discord"]["tools"]
def test_discord_server_not_in_core_tools(self):
from toolsets import _HERMES_CORE_TOOLS
assert "discord_server" not in _HERMES_CORE_TOOLS
def test_discord_server_not_in_other_toolsets(self):
from toolsets import TOOLSETS
for name, ts in TOOLSETS.items():
if name == "hermes-discord":
continue
# The gateway toolset might include it if it unions all platform tools
if name == "hermes-gateway":
continue
assert "discord_server" not in ts.get("tools", []), (
f"discord_server should not be in toolset '{name}'"
)
# ---------------------------------------------------------------------------
# Capability detection (privileged intents)
# ---------------------------------------------------------------------------
class TestCapabilityDetection:
def setup_method(self):
_reset_capability_cache()
def teardown_method(self):
_reset_capability_cache()
@patch("tools.discord_tool._discord_request")
def test_both_intents_enabled(self, mock_req):
# flags: GUILD_MEMBERS (1<<14) + MESSAGE_CONTENT (1<<18) = 278528
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
caps = _detect_capabilities("tok")
assert caps["has_members_intent"] is True
assert caps["has_message_content"] is True
assert caps["detected"] is True
@patch("tools.discord_tool._discord_request")
def test_no_intents(self, mock_req):
mock_req.return_value = {"flags": 0}
caps = _detect_capabilities("tok")
assert caps["has_members_intent"] is False
assert caps["has_message_content"] is False
assert caps["detected"] is True
@patch("tools.discord_tool._discord_request")
def test_limited_intent_variants_counted(self, mock_req):
# GUILD_MEMBERS_LIMITED (1<<15), MESSAGE_CONTENT_LIMITED (1<<19)
mock_req.return_value = {"flags": (1 << 15) | (1 << 19)}
caps = _detect_capabilities("tok")
assert caps["has_members_intent"] is True
assert caps["has_message_content"] is True
@patch("tools.discord_tool._discord_request")
def test_only_members_intent(self, mock_req):
mock_req.return_value = {"flags": 1 << 14}
caps = _detect_capabilities("tok")
assert caps["has_members_intent"] is True
assert caps["has_message_content"] is False
@patch("tools.discord_tool._discord_request")
def test_detection_failure_is_permissive(self, mock_req):
"""If detection fails (network/401/revoked token), expose everything
and let runtime errors surface. Silent failure should never hide
actions the bot actually has."""
mock_req.side_effect = DiscordAPIError(401, "unauthorized")
caps = _detect_capabilities("tok")
assert caps["detected"] is False
assert caps["has_members_intent"] is True
assert caps["has_message_content"] is True
@patch("tools.discord_tool._discord_request")
def test_detection_is_cached(self, mock_req):
mock_req.return_value = {"flags": 0}
_detect_capabilities("tok")
_detect_capabilities("tok")
_detect_capabilities("tok")
assert mock_req.call_count == 1
@patch("tools.discord_tool._discord_request")
def test_force_refresh(self, mock_req):
mock_req.return_value = {"flags": 0}
_detect_capabilities("tok")
_detect_capabilities("tok", force=True)
assert mock_req.call_count == 2
# ---------------------------------------------------------------------------
# Config allowlist
# ---------------------------------------------------------------------------
class TestConfigAllowlist:
def test_empty_string_returns_none(self, monkeypatch):
"""Empty config means no allowlist — all actions visible."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": ""}},
)
assert _load_allowed_actions_config() is None
def test_missing_key_returns_none(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {}},
)
assert _load_allowed_actions_config() is None
def test_comma_separated_string(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "list_guilds,list_channels,fetch_messages"}},
)
result = _load_allowed_actions_config()
assert result == ["list_guilds", "list_channels", "fetch_messages"]
def test_yaml_list(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": ["list_guilds", "server_info"]}},
)
result = _load_allowed_actions_config()
assert result == ["list_guilds", "server_info"]
def test_unknown_names_dropped(self, monkeypatch, caplog):
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "list_guilds,bogus_action,fetch_messages"}},
)
with caplog.at_level("WARNING"):
result = _load_allowed_actions_config()
assert result == ["list_guilds", "fetch_messages"]
assert "bogus_action" in caplog.text
def test_config_load_failure_is_permissive(self, monkeypatch):
"""If config can't be loaded at all, fall back to None (all allowed)."""
def bad_load():
raise RuntimeError("disk gone")
monkeypatch.setattr("hermes_cli.config.load_config", bad_load)
assert _load_allowed_actions_config() is None
def test_unexpected_type_ignored(self, monkeypatch, caplog):
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": {"unexpected": "dict"}}},
)
with caplog.at_level("WARNING"):
result = _load_allowed_actions_config()
assert result is None
assert "unexpected type" in caplog.text
# ---------------------------------------------------------------------------
# Action filtering combines intents + allowlist
# ---------------------------------------------------------------------------
class TestAvailableActions:
def test_all_available_when_unrestricted(self):
caps = {"detected": True, "has_members_intent": True, "has_message_content": True}
assert _available_actions(caps, None) == list(_ACTIONS.keys())
def test_no_members_intent_hides_member_actions(self):
caps = {"detected": True, "has_members_intent": False, "has_message_content": True}
actions = _available_actions(caps, None)
assert "search_members" not in actions
assert "member_info" not in actions
# fetch_messages stays — MESSAGE_CONTENT affects content field but action works
assert "fetch_messages" in actions
def test_no_message_content_keeps_fetch_messages(self):
"""MESSAGE_CONTENT affects the content field, not the action.
Hiding fetch_messages would lose author/timestamp/attachments access."""
caps = {"detected": True, "has_members_intent": True, "has_message_content": False}
actions = _available_actions(caps, None)
assert "fetch_messages" in actions
assert "list_pins" in actions
def test_allowlist_intersects_with_intents(self):
"""Allowlist can only narrow — not re-enable intent-gated actions."""
caps = {"detected": True, "has_members_intent": False, "has_message_content": True}
allowlist = ["list_guilds", "search_members", "fetch_messages"]
actions = _available_actions(caps, allowlist)
# search_members gated by intent → stripped even though allowlisted
assert actions == ["list_guilds", "fetch_messages"]
def test_empty_allowlist_yields_empty(self):
caps = {"detected": True, "has_members_intent": True, "has_message_content": True}
assert _available_actions(caps, []) == []
def test_allowlist_preserves_canonical_order(self):
caps = {"detected": True, "has_members_intent": True, "has_message_content": True}
# Pass allowlist out of canonical order
allowlist = ["fetch_messages", "list_guilds", "server_info"]
assert _available_actions(caps, allowlist) == ["list_guilds", "server_info", "fetch_messages"]
# ---------------------------------------------------------------------------
# Dynamic schema build (integration of intents + config)
# ---------------------------------------------------------------------------
class TestDynamicSchema:
def setup_method(self):
_reset_capability_cache()
def teardown_method(self):
_reset_capability_cache()
@patch("tools.discord_tool._discord_request")
def test_no_token_returns_none(self, mock_req, monkeypatch):
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
assert get_dynamic_schema() is None
mock_req.assert_not_called()
@patch("tools.discord_tool._discord_request")
def test_full_intents_full_schema(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": ""}},
)
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
schema = get_dynamic_schema()
actions = schema["parameters"]["properties"]["action"]["enum"]
assert set(actions) == set(_ACTIONS.keys())
# No content warning
assert "MESSAGE_CONTENT" not in schema["description"]
@patch("tools.discord_tool._discord_request")
def test_no_members_intent_removes_member_actions_from_schema(
self, mock_req, monkeypatch,
):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": ""}},
)
mock_req.return_value = {"flags": 1 << 18} # only MESSAGE_CONTENT
schema = get_dynamic_schema()
actions = schema["parameters"]["properties"]["action"]["enum"]
assert "search_members" not in actions
assert "member_info" not in actions
# Manifest description should also not advertise them
assert "search_members" not in schema["description"]
assert "member_info" not in schema["description"]
@patch("tools.discord_tool._discord_request")
def test_no_message_content_adds_warning_note(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": ""}},
)
mock_req.return_value = {"flags": 1 << 14} # only GUILD_MEMBERS
schema = get_dynamic_schema()
assert "MESSAGE_CONTENT" in schema["description"]
# But fetch_messages is still available
actions = schema["parameters"]["properties"]["action"]["enum"]
assert "fetch_messages" in actions
@patch("tools.discord_tool._discord_request")
def test_config_allowlist_narrows_schema(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "list_guilds,list_channels"}},
)
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
schema = get_dynamic_schema()
actions = schema["parameters"]["properties"]["action"]["enum"]
assert actions == ["list_guilds", "list_channels"]
# Manifest description should only show allowed ones (check for
# the signature marker, which is specific to manifest lines)
assert "list_guilds()" in schema["description"]
assert "add_role(" not in schema["description"]
assert "create_thread(" not in schema["description"]
@patch("tools.discord_tool._discord_request")
def test_empty_allowlist_with_valid_values_hides_tool(self, mock_req, monkeypatch):
"""If the allowlist resolves to zero valid actions (e.g. all names
were typos), get_dynamic_schema returns None so the tool is dropped
entirely rather than showing an empty enum."""
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "typo_one,typo_two"}},
)
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
assert get_dynamic_schema() is None
# ---------------------------------------------------------------------------
# Runtime allowlist enforcement (defense in depth — schema already filtered)
# ---------------------------------------------------------------------------
class TestRuntimeAllowlistEnforcement:
@patch("tools.discord_tool._discord_request")
def test_denied_action_blocked_at_runtime(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "list_guilds"}},
)
result = json.loads(discord_server(action="add_role", guild_id="1", user_id="2", role_id="3"))
assert "error" in result
assert "disabled by config" in result["error"]
mock_req.assert_not_called()
@patch("tools.discord_tool._discord_request")
def test_allowed_action_proceeds(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "list_guilds"}},
)
mock_req.return_value = []
result = json.loads(discord_server(action="list_guilds"))
assert "guilds" in result
# ---------------------------------------------------------------------------
# 403 enrichment
# ---------------------------------------------------------------------------
class Test403Enrichment:
def test_enrich_known_action(self):
msg = _enrich_403("add_role", '{"message":"Missing Permissions"}')
assert "MANAGE_ROLES" in msg
assert "Missing Permissions" in msg # Raw body preserved
def test_enrich_unknown_action_includes_body(self):
msg = _enrich_403("some_new_action", '{"message":"weird"}')
assert "some_new_action" in msg
assert "weird" in msg
@patch("tools.discord_tool._discord_request")
def test_403_in_runtime_is_enriched(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": ""}},
)
mock_req.side_effect = DiscordAPIError(403, '{"message":"Missing Permissions"}')
result = json.loads(discord_server(
action="add_role", guild_id="1", user_id="2", role_id="3",
))
assert "error" in result
assert "MANAGE_ROLES" in result["error"]
@patch("tools.discord_tool._discord_request")
def test_non_403_errors_are_not_enriched(self, mock_req, monkeypatch):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": ""}},
)
mock_req.side_effect = DiscordAPIError(500, "server error")
result = json.loads(discord_server(action="list_guilds"))
assert "500" in result["error"]
assert "MANAGE_ROLES" not in result["error"]
# ---------------------------------------------------------------------------
# model_tools integration — dynamic schema replaces static
# ---------------------------------------------------------------------------
class TestModelToolsIntegration:
def setup_method(self):
_reset_capability_cache()
def teardown_method(self):
_reset_capability_cache()
@patch("tools.discord_tool._discord_request")
def test_discord_server_schema_rebuilt_by_get_tool_definitions(
self, mock_req, monkeypatch,
):
"""When model_tools.get_tool_definitions runs with discord_server
available, it should replace the static schema with the dynamic one."""
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "list_guilds,server_info"}},
)
# Bot without GUILD_MEMBERS intent
mock_req.return_value = {"flags": 0}
from model_tools import get_tool_definitions
tools = get_tool_definitions(enabled_toolsets=["hermes-discord"], quiet_mode=True)
discord_tool = next(
(t for t in tools if t.get("function", {}).get("name") == "discord_server"),
None,
)
assert discord_tool is not None, "discord_server should be in the schema"
actions = discord_tool["function"]["parameters"]["properties"]["action"]["enum"]
assert actions == ["list_guilds", "server_info"]
@patch("tools.discord_tool._discord_request")
def test_discord_server_dropped_when_allowlist_empties_it(
self, mock_req, monkeypatch,
):
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"discord": {"server_actions": "all_bogus_names"}},
)
mock_req.return_value = {"flags": 0}
from model_tools import get_tool_definitions
tools = get_tool_definitions(enabled_toolsets=["hermes-discord"], quiet_mode=True)
names = [t.get("function", {}).get("name") for t in tools]
assert "discord_server" not in names

891
tools/discord_tool.py Normal file
View file

@ -0,0 +1,891 @@
"""Discord server introspection and management tool.
Provides the agent with the ability to interact with Discord servers
when running on the Discord gateway. Uses Discord REST API directly
with the bot token no dependency on the gateway adapter's client.
Only included in the hermes-discord toolset, so it has zero cost
for users on other platforms.
The schema exposed to the model is filtered by two gates:
1. Privileged intents detected from GET /applications/@me at schema
build time. Actions that require an intent the bot doesn't have
(search_members / member_info GUILD_MEMBERS intent) are hidden.
fetch_messages is kept regardless of MESSAGE_CONTENT intent, but
its description is annotated when the intent is missing.
2. User config allowlist at ``discord.server_actions``. If the user
sets a comma-separated list (or YAML list) of action names, only
those appear in the schema. Empty/unset means all intent-available
actions are exposed.
Per-guild permissions (MANAGE_ROLES etc.) are NOT pre-checked Discord
returns a 403 at call time and :func:`_enrich_403` maps it to
actionable guidance the model can relay to the user.
"""
import json
import logging
import os
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional, Tuple
from tools.registry import registry
logger = logging.getLogger(__name__)
DISCORD_API_BASE = "https://discord.com/api/v10"
# Application flag bits (from GET /applications/@me → "flags").
# Source: https://discord.com/developers/docs/resources/application#application-object-application-flags
_FLAG_GATEWAY_GUILD_MEMBERS = 1 << 14
_FLAG_GATEWAY_GUILD_MEMBERS_LIMITED = 1 << 15
_FLAG_GATEWAY_MESSAGE_CONTENT = 1 << 18
_FLAG_GATEWAY_MESSAGE_CONTENT_LIMITED = 1 << 19
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _get_bot_token() -> Optional[str]:
"""Resolve the Discord bot token from environment."""
return os.getenv("DISCORD_BOT_TOKEN", "").strip() or None
def _discord_request(
method: str,
path: str,
token: str,
params: Optional[Dict[str, str]] = None,
body: Optional[Dict[str, Any]] = None,
timeout: int = 15,
) -> Any:
"""Make a request to the Discord REST API."""
url = f"{DISCORD_API_BASE}{path}"
if params:
url += "?" + urllib.parse.urlencode(params)
data = None
if body is not None:
data = json.dumps(body).encode("utf-8")
req = urllib.request.Request(
url,
data=data,
method=method,
headers={
"Authorization": f"Bot {token}",
"Content-Type": "application/json",
"User-Agent": "Hermes-Agent (https://github.com/NousResearch/hermes-agent)",
},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
if resp.status == 204:
return None
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
error_body = ""
try:
error_body = e.read().decode("utf-8", errors="replace")
except Exception:
pass
raise DiscordAPIError(e.code, error_body) from e
class DiscordAPIError(Exception):
"""Raised when a Discord API call fails."""
def __init__(self, status: int, body: str):
self.status = status
self.body = body
super().__init__(f"Discord API error {status}: {body}")
# ---------------------------------------------------------------------------
# Channel type mapping
# ---------------------------------------------------------------------------
_CHANNEL_TYPE_NAMES = {
0: "text",
2: "voice",
4: "category",
5: "announcement",
10: "announcement_thread",
11: "public_thread",
12: "private_thread",
13: "stage",
15: "forum",
16: "media",
}
def _channel_type_name(type_id: int) -> str:
return _CHANNEL_TYPE_NAMES.get(type_id, f"unknown({type_id})")
# ---------------------------------------------------------------------------
# Capability detection (application intents)
# ---------------------------------------------------------------------------
# Module-level cache so the app/me endpoint is hit at most once per process.
_capability_cache: Optional[Dict[str, Any]] = None
def _detect_capabilities(token: str, *, force: bool = False) -> Dict[str, Any]:
"""Detect the bot's app-wide capabilities via GET /applications/@me.
Returns a dict with keys:
- ``has_members_intent``: GUILD_MEMBERS intent is enabled
- ``has_message_content``: MESSAGE_CONTENT intent is enabled
- ``detected``: detection succeeded (False means exposing everything
and letting runtime errors handle it)
Cached in a module-global. Pass ``force=True`` to re-fetch.
"""
global _capability_cache
if _capability_cache is not None and not force:
return _capability_cache
caps: Dict[str, Any] = {
"has_members_intent": True,
"has_message_content": True,
"detected": False,
}
try:
app = _discord_request("GET", "/applications/@me", token, timeout=5)
flags = int(app.get("flags", 0) or 0)
caps["has_members_intent"] = bool(
flags & (_FLAG_GATEWAY_GUILD_MEMBERS | _FLAG_GATEWAY_GUILD_MEMBERS_LIMITED)
)
caps["has_message_content"] = bool(
flags & (_FLAG_GATEWAY_MESSAGE_CONTENT | _FLAG_GATEWAY_MESSAGE_CONTENT_LIMITED)
)
caps["detected"] = True
except Exception as exc: # nosec — detection is best-effort
logger.info(
"Discord capability detection failed (%s); exposing all actions.", exc,
)
_capability_cache = caps
return caps
def _reset_capability_cache() -> None:
"""Test hook: clear the detection cache."""
global _capability_cache
_capability_cache = None
# ---------------------------------------------------------------------------
# Action implementations
# ---------------------------------------------------------------------------
def _list_guilds(token: str, **_kwargs: Any) -> str:
"""List all guilds the bot is a member of."""
guilds = _discord_request("GET", "/users/@me/guilds", token)
result = []
for g in guilds:
result.append({
"id": g["id"],
"name": g["name"],
"icon": g.get("icon"),
"owner": g.get("owner", False),
"permissions": g.get("permissions"),
})
return json.dumps({"guilds": result, "count": len(result)})
def _server_info(token: str, guild_id: str, **_kwargs: Any) -> str:
"""Get detailed information about a guild."""
g = _discord_request("GET", f"/guilds/{guild_id}", token, params={"with_counts": "true"})
return json.dumps({
"id": g["id"],
"name": g["name"],
"description": g.get("description"),
"icon": g.get("icon"),
"owner_id": g.get("owner_id"),
"member_count": g.get("approximate_member_count"),
"online_count": g.get("approximate_presence_count"),
"features": g.get("features", []),
"premium_tier": g.get("premium_tier"),
"premium_subscription_count": g.get("premium_subscription_count"),
"verification_level": g.get("verification_level"),
})
def _list_channels(token: str, guild_id: str, **_kwargs: Any) -> str:
"""List all channels in a guild, organized by category."""
channels = _discord_request("GET", f"/guilds/{guild_id}/channels", token)
# Organize: categories first, then channels under each
categories: Dict[Optional[str], Dict[str, Any]] = {}
uncategorized: List[Dict[str, Any]] = []
# First pass: collect categories
for ch in channels:
if ch["type"] == 4: # category
categories[ch["id"]] = {
"id": ch["id"],
"name": ch["name"],
"position": ch.get("position", 0),
"channels": [],
}
# Second pass: assign channels to categories
for ch in channels:
if ch["type"] == 4:
continue
entry = {
"id": ch["id"],
"name": ch.get("name", ""),
"type": _channel_type_name(ch["type"]),
"position": ch.get("position", 0),
"topic": ch.get("topic"),
"nsfw": ch.get("nsfw", False),
}
parent = ch.get("parent_id")
if parent and parent in categories:
categories[parent]["channels"].append(entry)
else:
uncategorized.append(entry)
# Sort
sorted_cats = sorted(categories.values(), key=lambda c: c["position"])
for cat in sorted_cats:
cat["channels"].sort(key=lambda c: c["position"])
uncategorized.sort(key=lambda c: c["position"])
result: List[Dict[str, Any]] = []
if uncategorized:
result.append({"category": None, "channels": uncategorized})
for cat in sorted_cats:
result.append({
"category": {"id": cat["id"], "name": cat["name"]},
"channels": cat["channels"],
})
total = sum(len(group["channels"]) for group in result)
return json.dumps({"channel_groups": result, "total_channels": total})
def _channel_info(token: str, channel_id: str, **_kwargs: Any) -> str:
"""Get detailed info about a specific channel."""
ch = _discord_request("GET", f"/channels/{channel_id}", token)
return json.dumps({
"id": ch["id"],
"name": ch.get("name"),
"type": _channel_type_name(ch["type"]),
"guild_id": ch.get("guild_id"),
"topic": ch.get("topic"),
"nsfw": ch.get("nsfw", False),
"position": ch.get("position"),
"parent_id": ch.get("parent_id"),
"rate_limit_per_user": ch.get("rate_limit_per_user", 0),
"last_message_id": ch.get("last_message_id"),
})
def _list_roles(token: str, guild_id: str, **_kwargs: Any) -> str:
"""List all roles in a guild."""
roles = _discord_request("GET", f"/guilds/{guild_id}/roles", token)
result = []
for r in sorted(roles, key=lambda r: r.get("position", 0), reverse=True):
result.append({
"id": r["id"],
"name": r["name"],
"color": f"#{r.get('color', 0):06x}" if r.get("color") else None,
"position": r.get("position", 0),
"mentionable": r.get("mentionable", False),
"managed": r.get("managed", False),
"member_count": r.get("member_count"),
"hoist": r.get("hoist", False),
})
return json.dumps({"roles": result, "count": len(result)})
def _member_info(token: str, guild_id: str, user_id: str, **_kwargs: Any) -> str:
"""Get info about a specific guild member."""
m = _discord_request("GET", f"/guilds/{guild_id}/members/{user_id}", token)
user = m.get("user", {})
return json.dumps({
"user_id": user.get("id"),
"username": user.get("username"),
"display_name": user.get("global_name"),
"nickname": m.get("nick"),
"avatar": user.get("avatar"),
"bot": user.get("bot", False),
"roles": m.get("roles", []),
"joined_at": m.get("joined_at"),
"premium_since": m.get("premium_since"),
})
def _search_members(token: str, guild_id: str, query: str, limit: int = 20, **_kwargs: Any) -> str:
"""Search for guild members by name."""
params = {"query": query, "limit": str(min(limit, 100))}
members = _discord_request("GET", f"/guilds/{guild_id}/members/search", token, params=params)
result = []
for m in members:
user = m.get("user", {})
result.append({
"user_id": user.get("id"),
"username": user.get("username"),
"display_name": user.get("global_name"),
"nickname": m.get("nick"),
"bot": user.get("bot", False),
"roles": m.get("roles", []),
})
return json.dumps({"members": result, "count": len(result)})
def _fetch_messages(
token: str, channel_id: str, limit: int = 50,
before: Optional[str] = None, after: Optional[str] = None,
**_kwargs: Any,
) -> str:
"""Fetch recent messages from a channel."""
params: Dict[str, str] = {"limit": str(min(limit, 100))}
if before:
params["before"] = before
if after:
params["after"] = after
messages = _discord_request("GET", f"/channels/{channel_id}/messages", token, params=params)
result = []
for msg in messages:
author = msg.get("author", {})
result.append({
"id": msg["id"],
"content": msg.get("content", ""),
"author": {
"id": author.get("id"),
"username": author.get("username"),
"display_name": author.get("global_name"),
"bot": author.get("bot", False),
},
"timestamp": msg.get("timestamp"),
"edited_timestamp": msg.get("edited_timestamp"),
"attachments": [
{"filename": a.get("filename"), "url": a.get("url"), "size": a.get("size")}
for a in msg.get("attachments", [])
],
"reactions": [
{"emoji": r.get("emoji", {}).get("name"), "count": r.get("count", 0)}
for r in msg.get("reactions", [])
] if msg.get("reactions") else [],
"pinned": msg.get("pinned", False),
})
return json.dumps({"messages": result, "count": len(result)})
def _list_pins(token: str, channel_id: str, **_kwargs: Any) -> str:
"""List pinned messages in a channel."""
messages = _discord_request("GET", f"/channels/{channel_id}/pins", token)
result = []
for msg in messages:
author = msg.get("author", {})
result.append({
"id": msg["id"],
"content": msg.get("content", "")[:200], # Truncate for overview
"author": author.get("username"),
"timestamp": msg.get("timestamp"),
})
return json.dumps({"pinned_messages": result, "count": len(result)})
def _pin_message(token: str, channel_id: str, message_id: str, **_kwargs: Any) -> str:
"""Pin a message in a channel."""
_discord_request("PUT", f"/channels/{channel_id}/pins/{message_id}", token)
return json.dumps({"success": True, "message": f"Message {message_id} pinned."})
def _unpin_message(token: str, channel_id: str, message_id: str, **_kwargs: Any) -> str:
"""Unpin a message from a channel."""
_discord_request("DELETE", f"/channels/{channel_id}/pins/{message_id}", token)
return json.dumps({"success": True, "message": f"Message {message_id} unpinned."})
def _create_thread(
token: str, channel_id: str, name: str,
message_id: Optional[str] = None,
auto_archive_duration: int = 1440,
**_kwargs: Any,
) -> str:
"""Create a thread in a channel."""
if message_id:
# Create thread from an existing message
path = f"/channels/{channel_id}/messages/{message_id}/threads"
body: Dict[str, Any] = {
"name": name,
"auto_archive_duration": auto_archive_duration,
}
else:
# Create a standalone thread
path = f"/channels/{channel_id}/threads"
body = {
"name": name,
"auto_archive_duration": auto_archive_duration,
"type": 11, # PUBLIC_THREAD
}
thread = _discord_request("POST", path, token, body=body)
return json.dumps({
"success": True,
"thread_id": thread["id"],
"name": thread.get("name"),
})
def _add_role(token: str, guild_id: str, user_id: str, role_id: str, **_kwargs: Any) -> str:
"""Add a role to a guild member."""
_discord_request("PUT", f"/guilds/{guild_id}/members/{user_id}/roles/{role_id}", token)
return json.dumps({"success": True, "message": f"Role {role_id} added to user {user_id}."})
def _remove_role(token: str, guild_id: str, user_id: str, role_id: str, **_kwargs: Any) -> str:
"""Remove a role from a guild member."""
_discord_request("DELETE", f"/guilds/{guild_id}/members/{user_id}/roles/{role_id}", token)
return json.dumps({"success": True, "message": f"Role {role_id} removed from user {user_id}."})
# ---------------------------------------------------------------------------
# Action dispatch + metadata
# ---------------------------------------------------------------------------
_ACTIONS = {
"list_guilds": _list_guilds,
"server_info": _server_info,
"list_channels": _list_channels,
"channel_info": _channel_info,
"list_roles": _list_roles,
"member_info": _member_info,
"search_members": _search_members,
"fetch_messages": _fetch_messages,
"list_pins": _list_pins,
"pin_message": _pin_message,
"unpin_message": _unpin_message,
"create_thread": _create_thread,
"add_role": _add_role,
"remove_role": _remove_role,
}
# Single-source-of-truth manifest: action → (signature, one-line description).
# Consumed by :func:`_build_schema` so the schema's top-level description
# always matches the registered action set.
_ACTION_MANIFEST: List[Tuple[str, str, str]] = [
("list_guilds", "()", "list servers the bot is in"),
("server_info", "(guild_id)", "server details + member counts"),
("list_channels", "(guild_id)", "all channels grouped by category"),
("channel_info", "(channel_id)", "single channel details"),
("list_roles", "(guild_id)", "roles sorted by position"),
("member_info", "(guild_id, user_id)", "lookup a specific member"),
("search_members", "(guild_id, query)", "find members by name prefix"),
("fetch_messages", "(channel_id)", "recent messages; optional before/after snowflakes"),
("list_pins", "(channel_id)", "pinned messages in a channel"),
("pin_message", "(channel_id, message_id)", "pin a message"),
("unpin_message", "(channel_id, message_id)", "unpin a message"),
("create_thread", "(channel_id, name)", "create a public thread; optional message_id anchor"),
("add_role", "(guild_id, user_id, role_id)", "assign a role"),
("remove_role", "(guild_id, user_id, role_id)", "remove a role"),
]
# Actions that require the GUILD_MEMBERS privileged intent.
_INTENT_GATED_MEMBERS = frozenset({"member_info", "search_members"})
# Per-action required params for runtime validation.
_REQUIRED_PARAMS: Dict[str, List[str]] = {
"server_info": ["guild_id"],
"list_channels": ["guild_id"],
"list_roles": ["guild_id"],
"member_info": ["guild_id", "user_id"],
"search_members": ["guild_id", "query"],
"channel_info": ["channel_id"],
"fetch_messages": ["channel_id"],
"list_pins": ["channel_id"],
"pin_message": ["channel_id", "message_id"],
"unpin_message": ["channel_id", "message_id"],
"create_thread": ["channel_id", "name"],
"add_role": ["guild_id", "user_id", "role_id"],
"remove_role": ["guild_id", "user_id", "role_id"],
}
# ---------------------------------------------------------------------------
# Config-based action allowlist
# ---------------------------------------------------------------------------
def _load_allowed_actions_config() -> Optional[List[str]]:
"""Read ``discord.server_actions`` from user config.
Returns a list of allowed action names, or ``None`` if the user
hasn't restricted the set (default: all actions allowed).
Accepts either a comma-separated string or a YAML list.
Unknown action names are dropped with a log warning.
"""
try:
from hermes_cli.config import load_config
cfg = load_config()
except Exception as exc:
logger.debug("discord_server: could not load config (%s); allowing all actions.", exc)
return None
raw = (cfg.get("discord") or {}).get("server_actions")
if raw is None or raw == "":
return None
if isinstance(raw, str):
names = [n.strip() for n in raw.split(",") if n.strip()]
elif isinstance(raw, (list, tuple)):
names = [str(n).strip() for n in raw if str(n).strip()]
else:
logger.warning(
"discord.server_actions: unexpected type %s; ignoring.", type(raw).__name__,
)
return None
valid = [n for n in names if n in _ACTIONS]
invalid = [n for n in names if n not in _ACTIONS]
if invalid:
logger.warning(
"discord.server_actions: unknown action(s) ignored: %s. "
"Known: %s",
", ".join(invalid), ", ".join(_ACTIONS.keys()),
)
return valid
def _available_actions(
caps: Dict[str, Any],
allowlist: Optional[List[str]],
) -> List[str]:
"""Compute the visible action list from intents + config allowlist.
Preserves the canonical order from :data:`_ACTIONS`.
"""
actions: List[str] = []
for name in _ACTIONS:
# Intent filter
if not caps.get("has_members_intent", True) and name in _INTENT_GATED_MEMBERS:
continue
# Config allowlist filter
if allowlist is not None and name not in allowlist:
continue
actions.append(name)
return actions
# ---------------------------------------------------------------------------
# Schema construction
# ---------------------------------------------------------------------------
def _build_schema(
actions: List[str],
caps: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Build the tool schema for the given filtered action list."""
caps = caps or {}
if not actions:
# Tool shouldn't be registered when empty, but guard anyway.
actions = list(_ACTIONS.keys())
# Action manifest lines (action-first, parameter-scoped).
manifest_lines = [
f" {name}{sig}{desc}"
for name, sig, desc in _ACTION_MANIFEST
if name in actions
]
manifest_block = "\n".join(manifest_lines)
content_note = ""
if caps.get("detected") and caps.get("has_message_content") is False:
content_note = (
"\n\nNOTE: Bot does NOT have the MESSAGE_CONTENT privileged intent. "
"fetch_messages and list_pins will return message metadata (author, "
"timestamps, attachments, reactions, pin state) but `content` will be "
"empty for messages not sent as a direct mention to the bot or in DMs. "
"Enable the intent in the Discord Developer Portal to see all content."
)
description = (
"Query and manage a Discord server via the REST API.\n\n"
"Available actions:\n"
f"{manifest_block}\n\n"
"Call list_guilds first to discover guild_ids, then list_channels for "
"channel_ids. Runtime errors will tell you if the bot lacks a specific "
"per-guild permission (e.g. MANAGE_ROLES for add_role)."
f"{content_note}"
)
properties: Dict[str, Any] = {
"action": {
"type": "string",
"enum": actions,
},
"guild_id": {
"type": "string",
"description": "Discord server (guild) ID.",
},
"channel_id": {
"type": "string",
"description": "Discord channel ID.",
},
"user_id": {
"type": "string",
"description": "Discord user ID.",
},
"role_id": {
"type": "string",
"description": "Discord role ID.",
},
"message_id": {
"type": "string",
"description": "Discord message ID.",
},
"query": {
"type": "string",
"description": "Member name prefix to search for (search_members).",
},
"name": {
"type": "string",
"description": "New thread name (create_thread).",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "Max results (default 50). Applies to fetch_messages, search_members.",
},
"before": {
"type": "string",
"description": "Snowflake ID for reverse pagination (fetch_messages).",
},
"after": {
"type": "string",
"description": "Snowflake ID for forward pagination (fetch_messages).",
},
"auto_archive_duration": {
"type": "integer",
"enum": [60, 1440, 4320, 10080],
"description": "Thread archive duration in minutes (create_thread, default 1440).",
},
}
return {
"name": "discord_server",
"description": description,
"parameters": {
"type": "object",
"properties": properties,
"required": ["action"],
},
}
def get_dynamic_schema() -> Optional[Dict[str, Any]]:
"""Return a schema filtered by current intents + config allowlist.
Called by ``model_tools.get_tool_definitions`` as a post-processing
step so the schema the model sees always reflects reality. Returns
``None`` when no actions are available (tool should be removed from
the schema list entirely).
"""
token = _get_bot_token()
if not token:
return None
caps = _detect_capabilities(token)
allowlist = _load_allowed_actions_config()
actions = _available_actions(caps, allowlist)
if not actions:
logger.warning(
"discord_server: config allowlist/intents left zero available actions; "
"hiding tool from this session."
)
return None
return _build_schema(actions, caps)
# ---------------------------------------------------------------------------
# 403 error enrichment
# ---------------------------------------------------------------------------
_ACTION_403_HINT = {
"pin_message": (
"Bot lacks MANAGE_MESSAGES permission in this channel. "
"Ask the server admin to grant the bot a role that has MANAGE_MESSAGES, "
"or a per-channel overwrite."
),
"unpin_message": (
"Bot lacks MANAGE_MESSAGES permission in this channel."
),
"create_thread": (
"Bot lacks CREATE_PUBLIC_THREADS in this channel, or cannot view it."
),
"add_role": (
"Either the bot lacks MANAGE_ROLES, or the target role sits higher "
"than the bot's highest role. Roles can only be assigned below the "
"bot's own position in the role hierarchy."
),
"remove_role": (
"Either the bot lacks MANAGE_ROLES, or the target role sits higher "
"than the bot's highest role."
),
"fetch_messages": (
"Bot cannot view this channel (missing VIEW_CHANNEL or READ_MESSAGE_HISTORY)."
),
"list_pins": (
"Bot cannot view this channel (missing VIEW_CHANNEL or READ_MESSAGE_HISTORY)."
),
"channel_info": (
"Bot cannot view this channel (missing VIEW_CHANNEL)."
),
"search_members": (
"Likely missing the Server Members privileged intent — enable it in the "
"Discord Developer Portal under your bot's settings."
),
"member_info": (
"Bot cannot see this guild member (missing Server Members intent or "
"insufficient permissions)."
),
}
def _enrich_403(action: str, body: str) -> str:
"""Return a user-friendly guidance string for a 403 on ``action``."""
hint = _ACTION_403_HINT.get(action)
base = f"Discord API 403 (forbidden) on '{action}'."
if hint:
return f"{base} {hint} (Raw: {body})"
return f"{base} (Raw: {body})"
# ---------------------------------------------------------------------------
# Check function
# ---------------------------------------------------------------------------
def check_discord_tool_requirements() -> bool:
"""Tool is available only when a Discord bot token is configured."""
return bool(_get_bot_token())
# ---------------------------------------------------------------------------
# Main handler
# ---------------------------------------------------------------------------
def discord_server(
action: str,
guild_id: str = "",
channel_id: str = "",
user_id: str = "",
role_id: str = "",
message_id: str = "",
query: str = "",
name: str = "",
limit: int = 50,
before: str = "",
after: str = "",
auto_archive_duration: int = 1440,
task_id: str = None,
) -> str:
"""Execute a Discord server action."""
token = _get_bot_token()
if not token:
return json.dumps({"error": "DISCORD_BOT_TOKEN not configured."})
action_fn = _ACTIONS.get(action)
if not action_fn:
return json.dumps({
"error": f"Unknown action: {action}",
"available_actions": list(_ACTIONS.keys()),
})
# Config-level allowlist gate (defense in depth — schema already filtered,
# but a stale cached schema from a prior config should not let denied
# actions through).
allowlist = _load_allowed_actions_config()
if allowlist is not None and action not in allowlist:
return json.dumps({
"error": (
f"Action '{action}' is disabled by config (discord.server_actions). "
f"Allowed: {', '.join(allowlist) if allowlist else '<none>'}"
),
})
local_vars = {
"guild_id": guild_id,
"channel_id": channel_id,
"user_id": user_id,
"role_id": role_id,
"message_id": message_id,
"query": query,
"name": name,
}
missing = [p for p in _REQUIRED_PARAMS.get(action, []) if not local_vars.get(p)]
if missing:
return json.dumps({
"error": f"Missing required parameters for '{action}': {', '.join(missing)}",
})
try:
return action_fn(
token=token,
guild_id=guild_id,
channel_id=channel_id,
user_id=user_id,
role_id=role_id,
message_id=message_id,
query=query,
name=name,
limit=limit,
before=before,
after=after,
auto_archive_duration=auto_archive_duration,
)
except DiscordAPIError as e:
logger.warning("Discord API error in action '%s': %s", action, e)
if e.status == 403:
return json.dumps({"error": _enrich_403(action, e.body)})
return json.dumps({"error": str(e)})
except Exception as e:
logger.exception("Unexpected error in discord_server action '%s'", action)
return json.dumps({"error": f"Unexpected error: {e}"})
# ---------------------------------------------------------------------------
# Tool registration
# ---------------------------------------------------------------------------
# Register with the full unfiltered schema. ``model_tools.get_tool_definitions``
# rebuilds this per-session via ``get_dynamic_schema`` so the model only ever
# sees intent-available, config-allowed actions. The static registration is a
# safe baseline for tools that inspect the registry directly.
_STATIC_SCHEMA = _build_schema(list(_ACTIONS.keys()), caps={"detected": False})
registry.register(
name="discord_server",
toolset="discord",
schema=_STATIC_SCHEMA,
handler=lambda args, **kw: discord_server(
action=args.get("action", ""),
guild_id=args.get("guild_id", ""),
channel_id=args.get("channel_id", ""),
user_id=args.get("user_id", ""),
role_id=args.get("role_id", ""),
message_id=args.get("message_id", ""),
query=args.get("query", ""),
name=args.get("name", ""),
limit=args.get("limit", 50),
before=args.get("before", ""),
after=args.get("after", ""),
auto_archive_duration=args.get("auto_archive_duration", 1440),
task_id=kw.get("task_id"),
),
check_fn=check_discord_tool_requirements,
requires_env=["DISCORD_BOT_TOKEN"],
)

View file

@ -304,7 +304,10 @@ TOOLSETS = {
"hermes-discord": { "hermes-discord": {
"description": "Discord bot toolset - full access (terminal has safety checks via dangerous command approval)", "description": "Discord bot toolset - full access (terminal has safety checks via dangerous command approval)",
"tools": _HERMES_CORE_TOOLS, "tools": _HERMES_CORE_TOOLS + [
# Discord server introspection & management (gated on DISCORD_BOT_TOKEN via check_fn)
"discord_server",
],
"includes": [] "includes": []
}, },