ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

feat(platforms): add typed plugin send paths

Browse Source 
Route plugin target parsing, validation, and host-driven delivery through PlatformEntry across CLI and cron while preserving the host-only send_message policy.
This commit is contained in:
Teknium
2026-08-08 12:59:16 -07:00
parent b58c7aaf9d
commit d409f67485
11 changed files with 530 additions and 543 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contributors/emails/itzhak.pan@gmail.com
+1
View File
@@ -0,0 +1 @@
yaozhen
contributors/emails/keviea@gmail.com
+1
View File
@@ -0,0 +1 @@
bka9
contributors/emails/markmnl@fmsg.io
+1
View File
@@ -0,0 +1 @@
markmnl
cron/scheduler.py
+15 -21
View File
@@ -1363,28 +1363,22 @@ def _resolve_single_delivery_target(job: dict, deliver_value: str) -> Optional[d
platform_name, rest = deliver_value.split(":", 1)
platform_key = platform_name.lower()
from tools.send_message_tool import _parse_target_ref
from tools.send_message_tool import (
prepare_send_message_platforms,
resolve_send_target,
)
parsed_chat_id, parsed_thread_id, is_explicit = _parse_target_ref(platform_key, rest)
if is_explicit:
chat_id, thread_id = parsed_chat_id, parsed_thread_id
else:
chat_id, thread_id = rest, None
# Resolve human-friendly labels like "Alice (dm)" to real IDs.
try:
from gateway.channel_directory import resolve_channel_name
resolved = resolve_channel_name(platform_key, chat_id)
if resolved:
parsed_chat_id, parsed_thread_id, resolved_is_explicit = _parse_target_ref(platform_key, resolved)
if resolved_is_explicit:
chat_id = parsed_chat_id
if parsed_thread_id is not None:
thread_id = parsed_thread_id
else:
chat_id = resolved
except Exception:
pass
prepare_send_message_platforms()
chat_id, thread_id, resolution_error = resolve_send_target(
platform_key, rest
)
if resolution_error:
logger.warning(
"Invalid cron delivery target '%s': %s",
deliver_value,
resolution_error,
)
return None
if (
thread_id is None
gateway/platform_registry.py
+14 -3
View File
@@ -173,10 +173,20 @@ class PlatformEntry:
# Signature:
# (target_ref: str) -> Optional[tuple[str, Optional[str]]]
#
# If the callable returns None the target proceeds to the usual directory
# resolution / verbatim fallback path.
# If the callable returns None the target proceeds to channel-directory
# resolution. No opaque fallback is applied.
parse_target_ref_fn: Optional[Callable[[str], Optional[tuple[str, Optional[str]]]]] = None
# Optional validation applied after parsing/normalization or
# channel-directory resolution. Return True to accept, False to reject, or
# a non-empty string to reject with that diagnostic.
validate_target_ref_fn: Optional[Callable[[str], bool | str]] = None
# Optional whole-request handler for custom platform delivery. Receives
# (args, normalized_chat_id, platform_name, pconfig) and may be sync/async.
# Prefer standalone_sender_fn when the standard send contract is enough.
send_message_handler: Optional[Callable[[dict, str, str, Any], Any]] = None
# ── Standalone (out-of-process) sending ──
# Optional: async coroutine that delivers a message without a live
# gateway adapter. Called by ``tools/send_message_tool._send_via_adapter``
@@ -286,7 +296,8 @@ class PlatformRegistry:
def unregister(self, name: str) -> bool:
"""Remove a platform entry. Returns True if it existed."""
self._deferred.pop(name, None)
return self._entries.pop(name, None) is not None
removed = self._entries.pop(name, None) is not None
return removed
def get(self, name: str) -> Optional[PlatformEntry]:
"""Look up a platform entry by name."""
hermes_cli/plugins.py
+8 -33
View File
@@ -1272,39 +1272,6 @@ class PluginContext:
self.manifest.name, provider.name,
)
# -- send_message enricher registration ---------------------------------
def register_send_message_enricher(
self,
platform_name: str,
handler: Callable,
schema_fragment: dict | None = None,
) -> None:
"""Register a send_message enricher for a plugin messaging platform.
Lets plugin platforms extend ``send_message`` with custom target
prefixes, schema fields, and send handlers without modifying core code.
Args:
platform_name: Platform key used in target strings
(e.g. ``"myplatform"`` → ``myplatform:chat_id``).
handler: Callable receiving ``(args, chat_id, platform_name, pconfig)``
and returning a dict like ``{"success": True, "message_id": "..."}``
or ``{"error": "..."}``. May be ``async def`` or a regular
function — the dispatcher detects coroutine functions via
``inspect.iscoroutinefunction`` and awaits as needed.
schema_fragment: Optional dict of JSON-schema properties to merge
into ``send_message``'s parameter schema so the LLM sees the
custom fields.
"""
from tools.send_message_tool import register_send_message_enricher as _register_enricher
_register_enricher(platform_name, handler, schema_fragment=schema_fragment)
logger.debug(
"Plugin %s registered send_message enricher: %s",
self.manifest.name,
platform_name,
)
# -- platform adapter registration ---------------------------------------
def register_platform(
@@ -1691,6 +1658,14 @@ class PluginManager:
self._discovered = True
return
if force:
# Remove concrete and deferred platform registrations before
# clearing ownership metadata. Otherwise disabled plugins (or a
# profile switch in a long-lived process) leak their parser or
# send handler into the next discovery pass.
from gateway.platform_registry import platform_registry
for platform_name in tuple(self._plugin_platform_names):
platform_registry.unregister(platform_name)
self._plugins.clear()
self._hooks.clear()
self._middleware.clear()
tests/tools/test_send_message_enrichers.py
-332
View File
@@ -1,332 +0,0 @@
"""Tests for send_message plugin enrichers."""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from tools.send_message_tool import (
SEND_MESSAGE_SCHEMA,
_SEND_MESSAGE_ENRICHERS,
_SEND_MESSAGE_SCHEMA_FRAGMENTS,
_parse_target_ref,
_send_to_platform,
get_send_message_schema,
register_send_message_enricher,
)
@pytest.fixture(autouse=True)
def _reset_enrichers():
"""Clear the enricher registry before and after every test."""
_SEND_MESSAGE_ENRICHERS.clear()
_SEND_MESSAGE_SCHEMA_FRAGMENTS.clear()
yield
_SEND_MESSAGE_ENRICHERS.clear()
_SEND_MESSAGE_SCHEMA_FRAGMENTS.clear()
class TestParseTargetRef:
def test_enricher_target_parsing(self):
"""Plugin-enriched platforms treat the target_ref as explicit."""
register_send_message_enricher("myplatform", AsyncMock())
chat_id, thread_id, is_explicit = _parse_target_ref("myplatform", "+123")
assert chat_id == "+123"
assert thread_id is None
assert is_explicit is True
def test_builtin_precedence(self):
"""Built-in platforms win over enrichers with the same name."""
register_send_message_enricher("telegram", AsyncMock())
chat_id, thread_id, is_explicit = _parse_target_ref("telegram", "777")
assert chat_id == "777"
assert thread_id is None
assert is_explicit is True
def test_enricher_empty_target_ref(self):
"""Empty target_ref for an enriched platform is not explicit."""
register_send_message_enricher("myplatform", AsyncMock())
chat_id, thread_id, is_explicit = _parse_target_ref("myplatform", "")
assert chat_id is None
assert thread_id is None
assert is_explicit is False
class TestPlatformEntryTargetParsing:
def test_parse_target_ref_fn_is_used(self):
"""PlatformEntry.parse_target_ref_fn is consulted for explicit targets."""
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "fmsg-parse-test"
def _parser(ref):
if ref.startswith("@") and "@" in ref[1:]:
return ref, None
return None
entry = PlatformEntry(
name=platform_name,
label="Fmsg parse test",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
parse_target_ref_fn=_parser,
)
platform_registry.register(entry)
try:
chat_id, thread_id, is_explicit = _parse_target_ref(
platform_name, "@alice@example.com"
)
assert chat_id == "@alice@example.com"
assert thread_id is None
assert is_explicit is True
finally:
platform_registry.unregister(platform_name)
def test_parse_target_ref_fn_returns_none_falls_through(self):
"""When parse_target_ref_fn returns None, target falls through to enricher."""
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "fmsg-fallback-test"
def _parser(ref):
return None
entry = PlatformEntry(
name=platform_name,
label="Fmsg fallback test",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
parse_target_ref_fn=_parser,
)
platform_registry.register(entry)
register_send_message_enricher(platform_name, AsyncMock())
try:
chat_id, thread_id, is_explicit = _parse_target_ref(
platform_name, "anything"
)
assert chat_id == "anything"
assert thread_id is None
assert is_explicit is True
finally:
platform_registry.unregister(platform_name)
def test_parse_target_ref_fn_thread_id(self):
"""parse_target_ref_fn can return a thread_id alongside chat_id."""
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "threaded-parse-test"
def _parser(ref):
if ":" in ref:
room, thread = ref.split(":", 1)
return room, thread
return None
entry = PlatformEntry(
name=platform_name,
label="Threaded parse test",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
parse_target_ref_fn=_parser,
)
platform_registry.register(entry)
try:
chat_id, thread_id, is_explicit = _parse_target_ref(
platform_name, "room-1:thread-99"
)
assert chat_id == "room-1"
assert thread_id == "thread-99"
assert is_explicit is True
finally:
platform_registry.unregister(platform_name)
def test_builtin_platform_ignores_parse_target_ref_fn(self):
"""Built-in platform parsers win over a rogue PlatformEntry parser."""
from gateway.platform_registry import PlatformEntry, platform_registry
def _parser(ref):
return "SHOULD_NOT_RETURN", None
entry = PlatformEntry(
name="telegram",
label="Telegram rogue",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
parse_target_ref_fn=_parser,
)
platform_registry.register(entry)
try:
chat_id, thread_id, is_explicit = _parse_target_ref("telegram", "12345")
assert chat_id == "12345"
assert thread_id is None
assert is_explicit is True
finally:
platform_registry.unregister("telegram")
class TestSchemaMerge:
def test_schema_fragment_stored(self):
"""Schema fragments are stored in _SEND_MESSAGE_SCHEMA_FRAGMENTS."""
register_send_message_enricher(
"myplatform",
AsyncMock(),
schema_fragment={"voice": {"type": "string", "description": "Voice setting"}},
)
assert "myplatform" in _SEND_MESSAGE_SCHEMA_FRAGMENTS
assert _SEND_MESSAGE_SCHEMA_FRAGMENTS["myplatform"]["voice"]["type"] == "string"
def test_get_send_message_schema_assembles_fragments(self):
"""get_send_message_schema returns a copy with fragments merged."""
register_send_message_enricher(
"myplatform",
AsyncMock(),
schema_fragment={"voice": {"type": "string", "description": "Voice setting"}},
)
schema = get_send_message_schema()
assert "voice" in schema["parameters"]["properties"]
assert schema["parameters"]["properties"]["voice"]["type"] == "string"
def test_original_schema_not_mutated(self):
"""The module-level SEND_MESSAGE_SCHEMA is never mutated."""
register_send_message_enricher(
"myplatform",
AsyncMock(),
schema_fragment={"voice": {"type": "string", "description": "Voice setting"}},
)
assert "voice" not in SEND_MESSAGE_SCHEMA["parameters"]["properties"]
def test_schema_fragment_without_registration(self):
"""No fragment is added when schema_fragment is omitted."""
register_send_message_enricher("myplatform", AsyncMock())
assert "myplatform" not in _SEND_MESSAGE_SCHEMA_FRAGMENTS
class TestHandlerRouting:
def test_async_handler_invoked(self):
"""The async enricher handler receives correct arguments."""
handler = AsyncMock(return_value={"success": True, "message_id": "msg_1"})
register_send_message_enricher("myplatform", handler)
pconfig = SimpleNamespace(enabled=True, token="tok", extra={})
result = asyncio.run(
_send_to_platform(
"myplatform",
pconfig,
"chat42",
"hello",
args={"target": "myplatform:chat42", "message": "hello", "intent": "greet"},
)
)
assert result == {"success": True, "message_id": "msg_1"}
handler.assert_awaited_once()
call_args = handler.await_args.args
assert call_args[0] == {"target": "myplatform:chat42", "message": "hello", "intent": "greet"}
assert call_args[1] == "chat42"
assert call_args[2] == "myplatform"
assert call_args[3] is pconfig
def test_sync_handler_invoked(self):
"""The sync enricher handler is called directly (not awaited)."""
handler = MagicMock(return_value={"success": True, "message_id": "msg_sync"})
register_send_message_enricher("myplatform", handler)
pconfig = SimpleNamespace(enabled=True, token="tok", extra={})
result = asyncio.run(
_send_to_platform(
"myplatform",
pconfig,
"chat42",
"hello",
args={"target": "myplatform:chat42", "message": "hello"},
)
)
assert result == {"success": True, "message_id": "msg_sync"}
handler.assert_called_once()
call_args = handler.call_args.args
assert call_args[0] == {"target": "myplatform:chat42", "message": "hello"}
assert call_args[1] == "chat42"
assert call_args[2] == "myplatform"
assert call_args[3] is pconfig
def test_async_handler_result(self):
"""Async enricher result is returned verbatim."""
handler = AsyncMock(return_value={"success": True, "message_id": "msg_1"})
register_send_message_enricher("myplatform", handler)
result = asyncio.run(
_send_to_platform(
"myplatform",
SimpleNamespace(enabled=True, token="tok", extra={}),
"chat42",
"hello",
)
)
assert result == {"success": True, "message_id": "msg_1"}
def test_sync_handler_result(self):
"""Sync enricher result is returned verbatim."""
handler = MagicMock(return_value={"success": True, "message_id": "msg_sync"})
register_send_message_enricher("myplatform", handler)
result = asyncio.run(
_send_to_platform(
"myplatform",
SimpleNamespace(enabled=True, token="tok", extra={}),
"chat42",
"hello",
)
)
assert result == {"success": True, "message_id": "msg_sync"}
def test_async_handler_error(self):
"""Async enricher exceptions are caught and surfaced as error dicts."""
handler = AsyncMock(side_effect=RuntimeError("boom"))
register_send_message_enricher("myplatform", handler)
result = asyncio.run(
_send_to_platform(
"myplatform",
SimpleNamespace(enabled=True, token="tok", extra={}),
"chat42",
"hello",
)
)
assert "error" in result
assert "boom" in result["error"]
def test_sync_handler_error(self):
"""Sync enricher exceptions are caught and surfaced as error dicts."""
handler = MagicMock(side_effect=RuntimeError("sync boom"))
register_send_message_enricher("myplatform", handler)
result = asyncio.run(
_send_to_platform(
"myplatform",
SimpleNamespace(enabled=True, token="tok", extra={}),
"chat42",
"hello",
)
)
assert "error" in result
assert "sync boom" in result["error"]
class TestFallback:
def test_no_enricher_falls_through(self):
"""Platforms without an enricher still reach _send_via_adapter."""
result = asyncio.run(
_send_to_platform(
"unknownplatform",
SimpleNamespace(enabled=True, token="tok", extra={}),
"chat42",
"hello",
)
)
# _send_via_adapter will error because no adapter is registered;
# the exact error text is not important, just that it didn't crash
# inside an enricher branch.
assert isinstance(result, dict)
assert "error" in result
tests/tools/test_send_message_plugin_extensibility.py
+271
View File
@@ -0,0 +1,271 @@
"""Cross-surface regressions for standalone platform send extensibility (#64900)."""
from __future__ import annotations
import asyncio
import json
import os
import subprocess
import sys
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from gateway.config import Platform
from gateway.platform_registry import PlatformEntry, platform_registry
from tools.send_message_tool import resolve_send_target, send_message_tool
@pytest.fixture
def plugin_platform():
name = "fmsg-ext-test"
seen: list[dict] = []
def parser(ref: str):
normalized = ref.strip().lower()
if normalized.startswith("@") and "@" in normalized[1:]:
return normalized, None
return None
def validator(ref: str):
return not ref.endswith("@blocked.example")
async def handler(args, chat_id, platform_name, pconfig):
seen.append({"args": dict(args), "chat_id": chat_id, "platform": platform_name})
return {"success": True, "platform": platform_name, "chat_id": chat_id}
entry = PlatformEntry(
name=name,
label="Fixture Message",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
parse_target_ref_fn=parser,
validate_target_ref_fn=validator,
send_message_handler=handler,
)
platform_registry.register(entry)
try:
yield name, entry, seen
finally:
platform_registry.unregister(name)
def _config_for(name: str):
platform = Platform(name)
pconfig = SimpleNamespace(enabled=True, token=None, extra={})
return platform, pconfig, SimpleNamespace(
platforms={platform: pconfig},
get_home_channel=lambda _platform: None,
)
def test_platform_parser_normalizes_and_validator_rejects(plugin_platform):
name, _entry, _seen = plugin_platform
assert resolve_send_target(name, " @Alice@Example.COM ") == (
"@alice@example.com",
None,
None,
)
chat_id, thread_id, error = resolve_send_target(
name, "@alice@blocked.example"
)
assert chat_id is None
assert thread_id is None
assert error == f"Invalid target '@alice@blocked.example' on {name}"
def test_registered_plugin_rejects_unrecognized_opaque_target(plugin_platform):
name, _entry, seen = plugin_platform
with patch("gateway.channel_directory.resolve_channel_name", return_value=None):
chat_id, thread_id, error = resolve_send_target(
name, "dm:opaque-recipient"
)
assert chat_id is None
assert thread_id is None
assert "plugin parser did not recognize it" in error
assert seen == []
def test_plugin_parser_failures_are_diagnosable_without_leaking_exception(plugin_platform):
name, entry, _seen = plugin_platform
def broken_parser(_ref):
raise RuntimeError("credential-shaped plugin detail")
entry.parse_target_ref_fn = broken_parser
assert resolve_send_target(name, "recipient") == (
None,
None,
f"Target parser failed for platform '{name}'",
)
entry.parse_target_ref_fn = lambda _ref: {"chat_id": "wrong-shape"}
assert resolve_send_target(name, "recipient") == (
None,
None,
f"Target parser for platform '{name}' returned an invalid result",
)
def test_plugin_validator_custom_diagnostic_blocks_delivery(plugin_platform):
name, entry, seen = plugin_platform
entry.validate_target_ref_fn = lambda _chat_id: "recipient is outside the allowlist"
with patch("gateway.channel_directory.resolve_channel_name", return_value=None):
chat_id, thread_id, error = resolve_send_target(name, "@alice@example.com")
assert chat_id is None
assert thread_id is None
assert error == (
f"Invalid target '@alice@example.com' on {name}: "
"recipient is outside the allowlist"
)
assert seen == []
@pytest.mark.parametrize("async_handler", [False, True])
def test_host_send_honors_sync_and_async_plugin_handlers(plugin_platform, async_handler):
name, entry, seen = plugin_platform
platform, pconfig, config = _config_for(name)
if not async_handler:
def sync_handler(args, chat_id, platform_name, pconfig):
seen.append({"args": dict(args), "chat_id": chat_id, "platform": platform_name})
return {"success": True, "platform": platform_name, "chat_id": chat_id}
entry.send_message_handler = sync_handler
with patch("gateway.config.load_gateway_config", return_value=config), \
patch("tools.interrupt.is_interrupted", return_value=False), \
patch("gateway.mirror.mirror_to_session", return_value=True):
result = json.loads(send_message_tool({
"target": f"{name}:@Alice@Example.COM",
"message": "hello",
"subject": "greeting",
}))
assert result["success"] is True
assert result["platform"] == name
assert result["chat_id"] == "@alice@example.com"
assert seen[-1]["args"]["subject"] == "greeting"
def test_cli_and_cron_share_plugin_target_normalization(plugin_platform, monkeypatch, capsys):
from cron.scheduler import _resolve_single_delivery_target
from hermes_cli.send_cmd import cmd_send
name, _entry, _seen = plugin_platform
_platform, _pconfig, config = _config_for(name)
args = SimpleNamespace(
list_targets=False,
to=f"{name}:@Alice@Example.COM",
message="hello",
file=None,
subject=None,
json=True,
quiet=False,
)
monkeypatch.setattr("hermes_cli.send_cmd._load_hermes_env", lambda: None)
with patch("gateway.config.load_gateway_config", return_value=config), \
patch("tools.interrupt.is_interrupted", return_value=False), \
patch("gateway.mirror.mirror_to_session", return_value=True), \
pytest.raises(SystemExit) as exc:
cmd_send(args)
assert exc.value.code == 0
assert json.loads(capsys.readouterr().out)["chat_id"] == "@alice@example.com"
cron_target = _resolve_single_delivery_target(
{"name": "fixture"}, f"{name}:@Alice@Example.COM"
)
assert cron_target == {
"platform": name,
"chat_id": "@alice@example.com",
"thread_id": None,
}
def test_send_message_remains_host_only(plugin_platform):
from tools.registry import registry
assert registry.get_entry("send_message") is None
def test_force_reload_unregisters_profile_owned_platform(plugin_platform, monkeypatch):
from hermes_cli.plugins import PluginManager
name, _entry, _seen = plugin_platform
manager = PluginManager()
manager._plugin_platform_names.add(name)
manager._discovered = True
monkeypatch.setattr(manager, "_discover_and_load_inner", lambda: None)
manager.discover_and_load(force=True)
assert platform_registry.get(name) is None
assert name not in manager._plugin_platform_names
def test_fresh_process_real_plugin_fixture_covers_host_send_and_cron(tmp_path):
"""A standalone directory plugin is visible to host-driven send paths."""
home = tmp_path / "home"
plugin = home / "plugins" / "fmsg-fixture"
plugin.mkdir(parents=True)
(plugin / "plugin.yaml").write_text(
"name: fmsg-fixture\nversion: 0.1.0\ndescription: fixture\nkind: platform\n"
)
(home / "config.yaml").write_text("plugins:\n enabled:\n - fmsg-fixture\n")
(plugin / "__init__.py").write_text(
"async def _send(args, chat_id, platform_name, pconfig):\n"
" return {'success': True, 'platform': platform_name, 'chat_id': chat_id}\n"
"def _parse(ref):\n"
" ref = ref.strip().lower()\n"
" return (ref, None) if ref.startswith('@') and '@' in ref[1:] else None\n"
"def register(ctx):\n"
" ctx.register_platform(name='fmsg', label='Fmsg', "
"adapter_factory=lambda cfg: None, check_fn=lambda: True, "
"parse_target_ref_fn=_parse, send_message_handler=_send)\n"
)
script = r'''
import json
from types import SimpleNamespace
from unittest.mock import patch
from hermes_cli.plugins import discover_plugins
from gateway.config import Platform
from tools.registry import registry
from tools.send_message_tool import send_message_tool
discover_plugins()
platform = Platform("fmsg")
pconfig = SimpleNamespace(enabled=True, token=None, extra={})
config = SimpleNamespace(platforms={platform: pconfig}, get_home_channel=lambda p: None)
with patch("gateway.config.load_gateway_config", return_value=config), \
patch("tools.interrupt.is_interrupted", return_value=False), \
patch("gateway.mirror.mirror_to_session", return_value=True):
host_send = json.loads(send_message_tool({"target": "fmsg:@Alice@Example.COM",
"message": "hello", "subject": "hi"}))
from cron.scheduler import _resolve_single_delivery_target
cron = _resolve_single_delivery_target({}, "fmsg:@Alice@Example.COM")
print(json.dumps({"host_send": host_send, "cron": cron,
"model_registered": registry.get_entry("send_message") is not None}))
'''
env = dict(os.environ)
env.update({
"HERMES_HOME": str(home),
"HERMES_KANBAN_TASK": "fixture",
"PYTHONPATH": os.getcwd(),
})
completed = subprocess.run(
[sys.executable, "-c", script],
cwd=os.getcwd(),
env=env,
text=True,
capture_output=True,
check=True,
)
payload = json.loads(completed.stdout.strip().splitlines()[-1])
assert payload["host_send"]["chat_id"] == "@alice@example.com"
assert payload["cron"]["chat_id"] == "@alice@example.com"
assert payload["model_registered"] is False
tests/tools/test_send_message_target_parse.py
+6 -14
View File
@@ -62,7 +62,6 @@ def test_send_message_routes_whatsapp_group_jid_without_home_fallback() -> None:
thread_id=None,
media_files=[],
force_document=False,
args={"action": "send", "target": "whatsapp:120363408391911677@g.us", "message": "hello group"},
)
@@ -117,11 +116,10 @@ def test_resolved_opaque_plugin_target_uses_directory_id() -> None:
thread_id=None,
media_files=[],
force_document=False,
args={"action": "send", "target": f"{platform_name}:Friendly name", "message": "hello"},
)
def test_unresolved_opaque_plugin_target_passes_through_verbatim() -> None:
def test_unresolved_plugin_target_requires_explicit_parser() -> None:
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "opaque-verbatim-test"
@@ -171,18 +169,12 @@ def test_unresolved_opaque_plugin_target_passes_through_verbatim() -> None:
finally:
platform_registry.unregister(platform_name)
assert result["success"] is True
assert result == {
"error": f"Could not resolve 'dm:panyaozhen' on {platform_name}. "
"The plugin parser did not recognize it and no channel-directory entry matched."
}
discover_mock.assert_called_once_with()
send_mock.assert_awaited_once_with(
platform,
pconfig,
"dm:panyaozhen",
"hello",
thread_id=None,
media_files=[],
force_document=False,
args={"action": "send", "target": f"{platform_name}:dm:panyaozhen", "message": "hello"},
)
send_mock.assert_not_awaited()
def test_unresolved_builtin_target_keeps_directory_error() -> None:
tools/send_message_tool.py
+155 -140
View File
@@ -11,7 +11,7 @@ import logging
import os
import re
import time
from typing import Any, Awaitable, Callable
from agent.redact import redact_sensitive_text
from agent.secret_scope import get_secret
@@ -86,52 +86,11 @@ _CAPTIONABLE_EXTS = _IMAGE_EXTS | _VIDEO_EXTS | {
_TELEGRAM_CAPTION_LIMIT = 1024
_DEFAULT_CAPTION_LIMIT = 4096
# ---------------------------------------------------------------------------
# Plugin enricher registry for send_message
# ---------------------------------------------------------------------------
def prepare_send_message_platforms() -> None:
"""Load enabled standalone plugins before tool schemas/cache keys are built."""
from hermes_cli.plugins import discover_plugins
SendMessageEnricher = Callable[[dict, str, str, Any], Awaitable[dict] | dict]
"""Callable receiving (args, chat_id, platform_name, pconfig) -> dict result.
May be sync or async — the dispatcher detects coroutine functions via
``inspect.iscoroutinefunction`` and awaits as needed.
"""
_SEND_MESSAGE_ENRICHERS: dict[str, SendMessageEnricher] = {}
"""platform_name -> enricher handler."""
_SEND_MESSAGE_SCHEMA_FRAGMENTS: dict[str, dict] = {}
"""platform_name -> schema fragment dict (JSON Schema properties)."""
def register_send_message_enricher(
platform_name: str,
handler: SendMessageEnricher,
schema_fragment: dict | None = None,
) -> None:
"""Register a platform-specific enricher for the send_message tool.
Enrichers let plugin platforms extend send_message with custom target
parsing, schema fields, and send logic without modifying core code.
"""
_SEND_MESSAGE_ENRICHERS[platform_name] = handler
if schema_fragment:
_SEND_MESSAGE_SCHEMA_FRAGMENTS[platform_name] = schema_fragment
def get_send_message_schema() -> dict:
"""Return a fresh copy of the send_message schema with plugin fragments merged.
Fragments are assembled on demand so deferred plugin registration never
mutates the shared ``SEND_MESSAGE_SCHEMA`` dict. Callers that need the
current wire schema (e.g. tool-search builders, MCP catalogues) should use
this instead of reading ``SEND_MESSAGE_SCHEMA`` directly.
"""
import copy
schema = copy.deepcopy(SEND_MESSAGE_SCHEMA)
for fragment in _SEND_MESSAGE_SCHEMA_FRAGMENTS.values():
schema["parameters"]["properties"].update(fragment)
return schema
discover_plugins()
def _media_caption_split(text, media_files, *, max_caption_len):
@@ -335,18 +294,13 @@ def _handle_react(args, remove=False):
platform_name = parts[0].strip().lower()
target_ref = parts[1].strip() if len(parts) > 1 else None
chat_id = None
prepare_send_message_platforms()
if target_ref:
chat_id, _thread_id, _ = _parse_target_ref(platform_name, target_ref)
if not chat_id:
try:
from gateway.channel_directory import resolve_channel_name
resolved = resolve_channel_name(platform_name, target_ref)
except Exception:
resolved = None
# Opaque platform-native ids (e.g. photon space GUIDs like
# 'any;-;+1555...') match no parser pattern and no directory
# entry — pass them through verbatim; the adapter validates.
chat_id = resolved or target_ref
chat_id, _thread_id, resolution_error = resolve_send_target(
platform_name, target_ref
)
if resolution_error:
return tool_error(resolution_error)
try:
from gateway.config import Platform, load_gateway_config
@@ -416,52 +370,13 @@ def _handle_send(args):
chat_id = None
thread_id = None
prepare_send_message_platforms()
if target_ref:
chat_id, thread_id, is_explicit = _parse_target_ref(platform_name, target_ref)
else:
is_explicit = False
# Resolve human-friendly channel names to numeric IDs
if target_ref and not is_explicit:
resolution_failed = False
try:
from gateway.channel_directory import resolve_channel_name
resolved = resolve_channel_name(platform_name, target_ref)
if resolved:
parsed_chat_id, parsed_thread_id, _ = _parse_target_ref(
platform_name, resolved
)
# Directory entries are trusted platform IDs. Preserve an
# opaque plugin ID even when no built-in parser recognizes it.
chat_id = parsed_chat_id or resolved
if parsed_thread_id is not None:
thread_id = parsed_thread_id
except Exception:
resolved = None
resolution_failed = True
if not resolved:
from gateway.config import Platform
from gateway.platform_registry import platform_registry
from hermes_cli.plugins import discover_plugins
discover_plugins()
entry = platform_registry.get(platform_name)
is_builtin = platform_name in {member.value for member in Platform}
if entry is not None and entry.source == "plugin" and not is_builtin:
# Registered plugin platforms may use opaque IDs unknown to
# core. Their adapter owns final target validation.
chat_id = target_ref
elif resolution_failed:
return tool_error(
f"Could not resolve '{target_ref}' on {platform_name}. "
f"Try using a numeric channel ID instead."
)
else:
return tool_error(
f"Could not resolve '{target_ref}' on {platform_name}. "
f"Use send_message(action='list') to see available targets."
)
chat_id, thread_id, resolution_error = resolve_send_target(
platform_name, target_ref
)
if resolution_error:
return tool_error(resolution_error)
from tools.interrupt import is_interrupted
if is_interrupted():
@@ -473,15 +388,18 @@ def _handle_send(args):
except Exception as e:
return json.dumps(_error(f"Failed to load gateway config: {e}"))
# Accept any platform name — built-in names resolve to their enum
# member, plugin platform names create dynamic members via _missing_().
from gateway.platform_registry import platform_registry
entry = platform_registry.get(platform_name)
is_builtin = platform_name in {member.value for member in Platform}
if not is_builtin and entry is None:
return tool_error(
f"Unknown or unregistered plugin platform: {platform_name}"
)
try:
platform = Platform(platform_name)
except (ValueError, KeyError):
if platform_name in _SEND_MESSAGE_ENRICHERS:
platform = platform_name
else:
return tool_error(f"Unknown platform: {platform_name}")
return tool_error(f"Unknown platform: {platform_name}")
pconfig = config.platforms.get(platform)
if not pconfig or not pconfig.enabled:
@@ -562,16 +480,22 @@ def _handle_send(args):
try:
from model_tools import _run_async
send_kwargs = {
"thread_id": thread_id,
"media_files": media_files,
"force_document": force_document_attachments,
}
# Preserve the exact built-in call contract; only custom handlers need
# the complete typed request.
if entry is not None and entry.send_message_handler is not None:
send_kwargs["args"] = args
result = _run_async(
_send_to_platform(
platform,
pconfig,
chat_id,
cleaned_message,
thread_id=thread_id,
media_files=media_files,
force_document=force_document_attachments,
args=args,
**send_kwargs,
)
)
if used_home_channel and isinstance(result, dict) and result.get("success"):
@@ -694,28 +618,118 @@ def _parse_target_ref(platform_name: str, target_ref: str):
if platform_name == "xmpp" and "@" in target_ref:
return target_ref, None, True
# Plugin platforms may register a custom target parser via PlatformEntry.
# This is evaluated before the blanket enricher fallback so plugins can
# opt individual target shapes in/out explicitly.
try:
from gateway.platform_registry import platform_registry
entry = platform_registry.get(platform_name)
if entry is not None and entry.parse_target_ref_fn is not None:
parsed = entry.parse_target_ref_fn(target_ref)
if parsed is not None:
chat_id, thread_id = parsed
return chat_id, thread_id, True
except Exception:
pass
# Plugin-enriched platforms (legacy blanket fallback)
if platform_name in _SEND_MESSAGE_ENRICHERS:
if target_ref:
return target_ref, None, True
return None, None, False
return None, None, False
def resolve_send_target(
platform_name: str, target_ref: str
) -> tuple[str | None, str | None, str | None]:
"""Resolve one send target identically for model/CLI/cron surfaces.
Channel-directory IDs are trusted. Plugin platforms must explicitly parse
native target syntax; unresolved strings never receive an opaque fallback.
The optional validator is the final authority over parser-normalized and
directory-resolved IDs.
"""
from gateway.config import Platform
from gateway.platform_registry import platform_registry
entry = platform_registry.get(platform_name)
def _validate(candidate: str) -> str | None:
if entry is None or entry.validate_target_ref_fn is None:
return None
try:
verdict = entry.validate_target_ref_fn(candidate)
except Exception:
logger.debug(
"Plugin target validator failed for %s", platform_name, exc_info=True
)
return f"Target validator failed for platform '{platform_name}'"
if verdict is True:
return None
if isinstance(verdict, str) and verdict:
return f"Invalid target '{target_ref}' on {platform_name}: {verdict}"
return f"Invalid target '{target_ref}' on {platform_name}"
if entry is not None and entry.parse_target_ref_fn is not None:
try:
parsed = entry.parse_target_ref_fn(target_ref)
except Exception:
logger.debug(
"Plugin target parser failed for %s", platform_name, exc_info=True
)
return None, None, f"Target parser failed for platform '{platform_name}'"
if parsed is not None:
if (
not isinstance(parsed, tuple)
or len(parsed) != 2
or not isinstance(parsed[0], str)
or not parsed[0]
or (parsed[1] is not None and not isinstance(parsed[1], str))
):
return (
None,
None,
f"Target parser for platform '{platform_name}' returned an invalid result",
)
parsed_chat_id, parsed_thread_id = parsed
error = _validate(parsed_chat_id)
return (None, None, error) if error else (
parsed_chat_id,
parsed_thread_id,
None,
)
parsed_chat_id, parsed_thread_id, explicit = _parse_target_ref(
platform_name, target_ref
)
if explicit and parsed_chat_id is not None:
error = _validate(parsed_chat_id)
return (None, None, error) if error else (
parsed_chat_id,
parsed_thread_id,
None,
)
resolution_failed = False
try:
from gateway.channel_directory import resolve_channel_name
resolved = resolve_channel_name(platform_name, target_ref)
except Exception:
resolved = None
resolution_failed = True
if resolved:
parsed_chat_id, parsed_thread_id, _ = _parse_target_ref(
platform_name, resolved
)
chat_id = parsed_chat_id or resolved
error = _validate(chat_id)
return (None, None, error) if error else (
chat_id,
parsed_thread_id,
None,
)
is_builtin = platform_name in {member.value for member in Platform}
if entry is None and not is_builtin:
return None, None, f"Unknown or unregistered plugin platform: {platform_name}"
if entry is not None and entry.source == "plugin" and not is_builtin:
return (
None,
None,
f"Could not resolve '{target_ref}' on {platform_name}. "
"The plugin parser did not recognize it and no channel-directory entry matched.",
)
hint = (
"Try using a numeric channel ID instead."
if resolution_failed
else "Use send_message(action='list') to see available targets."
)
return None, None, f"Could not resolve '{target_ref}' on {platform_name}. {hint}"
def _describe_media_for_mirror(media_files):
"""Return a human-readable mirror summary when a message only contains media."""
if not media_files:
@@ -1239,19 +1253,20 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
elif platform == Platform.YUANBAO:
result = await _send_yuanbao(chat_id, chunk)
else:
enricher = _SEND_MESSAGE_ENRICHERS.get(platform_name)
if enricher:
if args is None:
args = {}
from gateway.platform_registry import platform_registry
entry = platform_registry.get(platform_name)
handler = entry.send_message_handler if entry is not None else None
if handler is not None:
try:
import inspect
if inspect.iscoroutinefunction(enricher):
result = await enricher(args, chat_id, platform_name, pconfig)
else:
result = enricher(args, chat_id, platform_name, pconfig)
result = handler(args or {}, chat_id, platform_name, pconfig)
if inspect.isawaitable(result):
result = await result
return result
except Exception as e:
return {"error": f"Enricher send failed: {e}"}
return {"error": f"Plugin send_message handler failed: {e}"}
# Plugin platform: route through the gateway's live adapter if
# available, otherwise the plugin's standalone_sender_fn.
result = await _send_via_adapter(
website/docs/developer-guide/adding-platform-adapters.md
+58
View File
@@ -206,6 +206,64 @@ When you call `ctx.register_platform()`, the following integration points are ha
| Token lock (multi-profile) | Use `acquire_scoped_lock()` in your `connect()` |
| Orphaned config warning | Descriptive log when plugin is missing |
## Standalone send-path extensions
A standalone platform can participate in host-driven outbound delivery through
direct `hermes send --to ...` and cron `deliver=platform:...` by declaring send
behavior on the same `PlatformEntry` created by `ctx.register_platform()`.
`send_message` is intentionally not an agent-callable model tool; plugins must
not register an equivalent model surface that lets the agent initiate outbound
messages on its own.
```python
async def _send_request(args, chat_id, platform_name, pconfig):
# `args` contains the host-driven send request fields.
message_id = await client.send(
address=chat_id,
body=args["message"],
subject=args.get("subject"),
)
return {"success": True, "platform": platform_name,
"chat_id": chat_id, "message_id": message_id}
def _parse_address(raw):
normalized = raw.strip().lower()
if normalized.startswith("@") and "@" in normalized[1:]:
return normalized, None # (chat_id, optional thread_id)
return None # continue to channel-directory resolution
def _validate_address(address):
# True accepts; False rejects; a string rejects with that diagnostic.
return True if address.endswith("@example.com") else "unsupported domain"
def register(ctx):
ctx.register_platform(
name="fmsg",
label="Fixture Message",
adapter_factory=lambda cfg: FmsgAdapter(cfg),
check_fn=check_requirements,
parse_target_ref_fn=_parse_address,
validate_target_ref_fn=_validate_address,
# May be a regular function or async def. Hermes awaits any awaitable
# result, including callable objects and functools.partial wrappers.
send_message_handler=_send_request,
# Prefer this lower-level hook when cron must send from a process
# without the live gateway.
standalone_sender_fn=_standalone_send,
)
```
Target resolution is shared across all three outbound surfaces. Parser output
is normalized first and channel-directory IDs are trusted. A plugin parser must
explicitly accept native target syntax; unresolved strings are never passed
through opaquely. Unknown platforms and validator failures return a diagnostic
instead of silently attempting delivery. Plugin force-reload/profile
transitions unregister owned entries, so parsers and handlers cannot leak into
the next profile.
## Env-Driven Auto-Configuration
Most users set up a platform by dropping env vars into `~/.hermes/.env` rather than editing `config.yaml`. The `env_enablement_fn` hook lets your plugin pick those env vars up **before** the adapter is constructed, so `hermes gateway status`, `get_connected_platforms()`, and cron delivery see the correct state without instantiating the platform SDK.