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

Merge branch 'simp/r3-33-D' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:41:26 -07:00
parent 0615725d45 37ae28f187
commit 8d77967441
4 changed files with 520 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/react_to_message_tool.py
+16 -23
View File
@@ -1,12 +1,12 @@
#!/usr/bin/env python3
"""Let the agent react to a message with an emoji in the Hermes desktop app.
The conversational counterpart to the user's tapback: same reaction store,
same one-per-author semantics, written with ``author="agent"``. Lives in the
``desktop_ui`` toolset so it costs nothing on other surfaces (platform adapters
expose reactions via ``send_message(action="react")``). Defaults to the message
that triggered this turn so the model needn't thread row ids through tool calls,
and emits ``message.reaction`` so the renderer paints it without a resume.
The conversational counterpart to the user's tapback: same reaction store, same
one-per-author semantics, written with ``author="agent"``. Lives in the ``desktop_ui``
toolset so it costs nothing on other surfaces (platform adapters expose reactions via
``send_message(action="react")``). Defaults to the message that triggered this turn so
the model needn't thread row ids through tool calls, and emits ``message.reaction`` so
the renderer paints it without a resume.
"""
import json
@@ -30,15 +30,14 @@ def _react(emoji: str, message_row_id, messages_back, *, db, session_key: str) -
row_id = message_row_id
target_role = "user"
if row_id is None:
# Default target: the latest user message. `messages_back` steps to earlier
# user turns (1 = the one before) — ids aren't visible to the model and
# quoting text is ambiguous, but "two messages ago" is how a person thinks.
# Default target: the latest user message. `messages_back` steps to earlier user
# turns (1 = the one before) — ids aren't visible to the model and quoting text is
# ambiguous, but "two messages ago" is how a person thinks.
back = max(0, int(messages_back or 0))
row_id = db.latest_message_row_id(session_key, role="user", offset=back)
if row_id is None:
return tool_error(
f"No user message found {back} back." if back else "No user message to react to yet."
)
f"No user message found {back} back." if back else "No user message to react to yet.")
else:
target_role = db.get_message_role(session_key, int(row_id)) or "user"
@@ -49,28 +48,22 @@ def _react(emoji: str, message_row_id, messages_back, *, db, session_key: str) -
if reactions is None:
return tool_error(f"Message {row_id} is not part of this conversation.")
# Paint it live. A missing bridge (non-desktop surface) is not an error — the
# reaction is persisted and shows on next load. `role` lets the renderer match
# a live message that doesn't know its durable row id yet (learned on resume).
# Paint it live. A missing bridge (non-desktop surface) is not an error — the reaction
# is persisted and shows on next load. `role` lets the renderer match a live message
# that doesn't know its durable row id yet (learned on resume).
try:
desktop_ui.emit(
"message.reaction",
{"row_id": int(row_id), "reactions": reactions, "role": target_role},
)
"message.reaction", {"row_id": int(row_id), "reactions": reactions, "role": target_role})
except Exception:
pass
return json.dumps(
{"success": True, "row_id": int(row_id), "reactions": reactions}, ensure_ascii=False
)
return json.dumps({"success": True, "row_id": int(row_id), "reactions": reactions}, ensure_ascii=False)
def react_to_message_tool(emoji: str, message_row_id=None, messages_back=None) -> str:
"""Attach (or with an empty ``emoji`` retract) the agent's reaction."""
emoji = (emoji or "").strip()
session_key = get_session_env("HERMES_SESSION_KEY", "") or get_session_env(
"HERMES_SESSION_ID", ""
)
session_key = get_session_env("HERMES_SESSION_KEY", "") or get_session_env("HERMES_SESSION_ID", "")
if not session_key:
return tool_error("No active session — reactions need a persisted conversation.")
tools/send_message_senders.py
+226 -338
View File
@@ -14,67 +14,52 @@ _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".gif"}
_VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".3gp"}
_AUDIO_EXTS = {".ogg", ".opus", ".mp3", ".m2a", ".wav", ".m4a", ".flac"}
_VOICE_EXTS = {".ogg", ".opus"}
# Telegram's Bot API sendAudio only accepts MP3 / M4A. Other audio
# formats either route through sendVoice (Opus/OGG) or fall back to
# document delivery.
# Telegram's sendAudio only accepts MP3 / M4A; other audio goes via sendVoice
# (Opus/OGG) or falls back to document delivery.
_TELEGRAM_SEND_AUDIO_EXTS = {".mp3", ".m4a"}
# Extensions that carry a native caption on the media bubble itself
# (photo/video/document). Voice/audio notes are excluded: a caption on a
# voice note reads as a separate label rather than a bubble caption, and the
# established convention is to keep the accompanying text as its own message.
_CAPTIONABLE_EXTS = _IMAGE_EXTS | _VIDEO_EXTS | {
".pdf", ".doc", ".docx", ".txt", ".md", ".csv", ".xlsx", ".zip",
}
# Extensions that carry a native caption on the media bubble (photo/video/document).
# Voice/audio notes are excluded: a caption on a voice note reads as a separate label,
# so the accompanying text stays its own message.
_CAPTIONABLE_EXTS = _IMAGE_EXTS | _VIDEO_EXTS | {".pdf", ".doc", ".docx", ".txt", ".md", ".csv", ".xlsx", ".zip"}
# Per-platform native caption length limits (characters). Text longer than
# the limit can't ride on the media bubble and stays a separate body message.
# Telegram's photo/video caption cap is 1024; WhatsApp and Discord are far
# more generous, so a conservative shared ceiling keeps behavior predictable.
# Native caption length limits (characters). Telegram's photo/video cap is 1024;
# WhatsApp/Discord are far more generous, so one conservative shared ceiling.
_TELEGRAM_CAPTION_LIMIT = 1024
_DEFAULT_CAPTION_LIMIT = 4096
def _media_caption_split(text, media_files, *, max_caption_len):
"""Decide whether the accompanying text rides on the media bubble.
"""Single chokepoint deciding whether text rides on the media bubble as its caption.
Single chokepoint for ``MEDIA:<path> caption`` across every sender.
Returns ``(caption, "")`` — text becomes the native caption, no separate
body message — only for exactly one media file of a captionable kind
(image/video/document, not a voice/audio note) whose text fits
``max_caption_len``. Otherwise ``(None, text)``: text goes as a separate
message (multi-file caption→file association is ambiguous).
Length is measured in codepoints: never under-counts Telegram's UTF-16
units for BMP text, so over-counting only fails safe. The Telegram sender
re-checks the *formatted* caption since escaping inflates it.
``(caption, "")`` only for exactly one captionable file (not a voice/audio note)
whose text fits ``max_caption_len``; otherwise ``(None, text)`` — multi-file
caption→file association is ambiguous. Length is codepoints, which never
under-counts Telegram's UTF-16 units for BMP text (over-counting fails safe);
the Telegram sender re-checks the *formatted* caption since escaping inflates it.
"""
stripped = (text or "").strip()
media = media_files or []
if not stripped or len(media) != 1:
if not stripped or len(media) != 1 or len(stripped) > max_caption_len:
return None, text
media_path, is_voice = media[0]
if is_voice or os.path.splitext(media_path)[1].lower() not in _CAPTIONABLE_EXTS:
return None, text
if len(stripped) > max_caption_len:
return None, text
return stripped, ""
_URL_SECRET_QUERY_RE = re.compile(
r"([?&](?:access_token|api[_-]?key|auth[_-]?token|token|signature|sig)=)([^&#\s]+)",
re.IGNORECASE,
)
re.IGNORECASE)
_GENERIC_SECRET_ASSIGN_RE = re.compile(
r"\b(access_token|api[_-]?key|auth[_-]?token|signature|sig)\s*=\s*([^\s,;]+)",
re.IGNORECASE,
)
r"\b(access_token|api[_-]?key|auth[_-]?token|signature|sig)\s*=\s*([^\s,;]+)", re.IGNORECASE)
def _sanitize_error_text(text) -> str:
"""Redact secrets from error text before surfacing it to users/models."""
redacted = redact_sensitive_text(text)
redacted = _URL_SECRET_QUERY_RE.sub(lambda m: f"{m.group(1)}***", redacted)
redacted = _GENERIC_SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}=***", redacted)
return redacted
return _GENERIC_SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}=***", redacted)
def _error(message: str) -> dict:
@@ -82,6 +67,14 @@ def _error(message: str) -> dict:
return {"error": _sanitize_error_text(message)}
def _success(platform: str, chat_id, warnings=None, **fields) -> dict:
"""Standard success payload; ``warnings`` is only included when non-empty."""
result = {"success": True, "platform": platform, "chat_id": chat_id, **fields}
if warnings:
result["warnings"] = warnings
return result
def _display_chat_id(platform_name: str, chat_id: str) -> str:
"""Return a result-safe chat identifier for tool transcripts/log consumers."""
if platform_name == "signal" and str(chat_id).startswith("group:"):
@@ -89,16 +82,16 @@ def _display_chat_id(platform_name: str, chat_id: str) -> str:
return chat_id
_NO_DELIVERABLE = "No deliverable text or media remained after processing MEDIA tags"
_TELEGRAM_TRANSIENT_MARKERS = (
"bad gateway", "502", "too many requests", "429",
"service unavailable", "503", "gateway timeout", "504",
)
"service unavailable", "503", "gateway timeout", "504")
def _telegram_retry_delay(exc: Exception, attempt: int) -> float | None:
"""Seconds to wait before retrying, or None when the error is final.
Honours Telegram's ``retry_after``; timeouts are never retried (the send
may have gone through); 5xx/429 back off exponentially."""
"""Seconds to wait before retrying, or None when final. Honours ``retry_after``;
timeouts are never retried (the send may have gone through); 5xx/429 back off."""
retry_after = getattr(exc, "retry_after", None)
if retry_after is not None:
try:
@@ -125,11 +118,7 @@ async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs)
raise
logger.warning(
"Transient Telegram send failure (attempt %d/%d), retrying in %.1fs: %s",
attempt + 1,
attempts,
delay,
_sanitize_error_text(exc),
)
attempt + 1, attempts, delay, _sanitize_error_text(exc))
await asyncio.sleep(delay)
@@ -139,9 +128,8 @@ def _is_telegram_thread_not_found(error: Exception) -> bool:
def _telegram_bot(token):
"""Build a Bot honouring TELEGRAM_PROXY (config ``telegram.proxy_url``);
without it the standalone path bypasses the proxy and times out where
api.telegram.org is blocked. Falls back to a direct connection."""
"""Bot honouring TELEGRAM_PROXY (``telegram.proxy_url``) — without it the standalone
path times out where api.telegram.org is blocked. Falls back to a direct connection."""
from telegram import Bot
try:
@@ -153,20 +141,16 @@ def _telegram_bot(token):
try:
from telegram.request import HTTPXRequest
logger.info("send_message: standalone Telegram send routed through proxy %s", proxy)
return Bot(
token=token,
request=HTTPXRequest(proxy=proxy),
get_updates_request=HTTPXRequest(proxy=proxy),
)
return Bot(token=token, request=HTTPXRequest(proxy=proxy),
get_updates_request=HTTPXRequest(proxy=proxy))
except Exception as proxy_err:
logger.warning("send_message: failed to attach Telegram proxy (%s), falling back to direct connection", proxy_err)
return Bot(token=token)
def _telegram_thread_kwargs(thread_id):
"""Map a topic id to ``message_thread_id`` kwargs. Forum "General" is
thread "1" on incoming updates but Bot API rejects message_thread_id=1
("Message thread not found"), so it maps to no thread — same as the adapter."""
"""Topic id -> ``message_thread_id`` kwargs. Forum "General" is thread "1" inbound but
the Bot API rejects message_thread_id=1, so it maps to no thread — same as the adapter."""
if thread_id is None:
return {}
try:
@@ -188,8 +172,8 @@ def _strip_mdv2_safe(text):
async def _telegram_send_media(bot, chat_id, f, ext, is_voice, force_document, **kwargs):
"""Pick the Bot API media method by extension: photo (unless forced to
document), video, voice note, sendAudio (MP3/M4A only), else document."""
"""Bot API media method by extension: photo (unless forced document), video, voice
note, sendAudio (MP3/M4A only), else document."""
if ext in _IMAGE_EXTS and not force_document:
return await bot.send_photo(chat_id=chat_id, photo=f, **kwargs)
if ext in _VIDEO_EXTS:
@@ -202,43 +186,37 @@ async def _telegram_send_media(bot, chat_id, f, ext, is_voice, force_document, *
async def _telegram_send_text_chunk(bot, chat_id, chunk, parse_mode, has_html, text_kwargs):
"""Send one formatted text chunk with the adapter-matching fallbacks:
thread-not-found -> retry without ``message_thread_id`` (dropped from
``text_kwargs`` for later chunks too); parse failure -> plain text."""
try:
"""One formatted text chunk with adapter-matching fallbacks: thread-not-found -> retry
without ``message_thread_id`` (dropped from ``text_kwargs`` for later chunks too);
parse failure -> plain text."""
async def send(text, mode):
return await _send_telegram_message_with_retry(
bot, chat_id=chat_id, text=chunk, parse_mode=parse_mode, **text_kwargs
)
bot, chat_id=chat_id, text=text, parse_mode=mode, **text_kwargs)
try:
return await send(chunk, parse_mode)
except Exception as md_error:
if _is_telegram_thread_not_found(md_error) and text_kwargs.get("message_thread_id") is not None:
logger.warning(
"Thread %s not found in _send_telegram, retrying without message_thread_id",
text_kwargs.get("message_thread_id"),
)
text_kwargs.get("message_thread_id"))
text_kwargs.pop("message_thread_id", None)
return await _send_telegram_message_with_retry(
bot, chat_id=chat_id, text=chunk, parse_mode=parse_mode, **text_kwargs
)
return await send(chunk, parse_mode)
err_text = str(md_error).lower()
if "parse" in err_text or "markdown" in err_text or "html" in err_text:
logger.warning(
"Parse mode %s failed in _send_telegram, falling back to plain text: %s",
parse_mode,
_sanitize_error_text(md_error),
)
plain = chunk if has_html else _strip_mdv2_safe(chunk)
return await _send_telegram_message_with_retry(
bot, chat_id=chat_id, text=plain, parse_mode=None, **text_kwargs
)
parse_mode, _sanitize_error_text(md_error))
return await send(chunk if has_html else _strip_mdv2_safe(chunk), None)
raise
async def _telegram_send_one_media(
bot, chat_id, media_path, is_voice, *, caption, parse_mode, has_html, thread_kwargs, force_document
):
"""Upload one file with the adapter-matching fallbacks (thread-not-found ->
no ``message_thread_id``; caption parse failure -> plain caption). Each
retry re-seeks the file because the first attempt consumed it."""
"""Upload one file with adapter-matching fallbacks (thread-not-found -> no
``message_thread_id``; caption parse failure -> plain caption). Retries re-seek
the file because the first attempt consumed it."""
ext = os.path.splitext(media_path)[1].lower()
voice_note = ext in _VOICE_EXTS and is_voice
with open(media_path, "rb") as f:
@@ -266,8 +244,7 @@ async def _telegram_send_one_media(
if _is_telegram_thread_not_found(media_err) and media_kwargs.get("message_thread_id"):
logger.warning(
"Thread %s not found for media send, retrying without message_thread_id",
media_kwargs["message_thread_id"],
)
media_kwargs["message_thread_id"])
f.seek(0)
media_kwargs.pop("message_thread_id", None)
return await _send(**media_kwargs)
@@ -275,8 +252,7 @@ async def _telegram_send_one_media(
if media_kwargs.get("parse_mode") and ("parse" in err_text or "caption" in err_text):
logger.warning(
"Caption parse failed for media send, retrying plain: %s",
_sanitize_error_text(media_err),
)
_sanitize_error_text(media_err))
f.seek(0)
media_kwargs.pop("parse_mode", None)
if not has_html and media_kwargs.get("caption"):
@@ -286,17 +262,12 @@ async def _telegram_send_one_media(
async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False, force_document=False):
"""Send via Telegram Bot API (one-shot, no polling needed).
Markdown is converted to MarkdownV2 via the gateway adapter's
``format_message`` so bold/links/headers render; a message that already
contains HTML tags skips that and is sent with ``parse_mode='HTML'``.
Parse failures fall back to plain text so the message still delivers.
"""
"""One-shot Telegram Bot API send. Markdown -> MarkdownV2 via the adapter's
``format_message``; text already containing HTML tags is sent as HTML instead.
Parse failures fall back to plain text so the message still delivers."""
try:
from telegram.constants import ParseMode
# Auto-detect HTML tags: if present, skip MarkdownV2 and send as HTML.
_has_html = bool(re.search(r'<[a-zA-Z/][^>]*>', message))
if _has_html:
formatted = message
@@ -324,75 +295,55 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No
last_msg = None
warnings = []
# MEDIA caption: a single captionable file + short text rides on the
# bubble as its *formatted* caption. Formatting can inflate a raw
# <1024 string past Telegram's 1024 cap, so re-check in UTF-16 units
# and fall back to a separate body message.
# MEDIA caption: a single captionable file + short text rides on the bubble as
# its *formatted* caption. Formatting can inflate a raw <1024 string past
# Telegram's cap, so re-check in UTF-16 units and fall back to a separate body.
_tg_caption = None
from gateway.platforms.base import BasePlatformAdapter, utf16_len
_cap, _ = _media_caption_split(
message, media_files, max_caption_len=_TELEGRAM_CAPTION_LIMIT
)
_cap, _ = _media_caption_split(message, media_files, max_caption_len=_TELEGRAM_CAPTION_LIMIT)
if _cap is not None and utf16_len(formatted) <= _TELEGRAM_CAPTION_LIMIT:
_tg_caption = formatted
formatted = "" # suppress the separate text send below
if formatted.strip():
# Chunk *after* formatting, in UTF-16 units: MarkdownV2/HTML
# escaping inflates text, so a raw-<4096 message can exceed the
# limit once formatted and be rejected as "Message is too long".
# Chunk *after* formatting, in UTF-16 units: MarkdownV2/HTML escaping inflates
# text, so a raw-<4096 message can exceed the limit once formatted.
for chunk in BasePlatformAdapter.truncate_message(formatted, 4096, len_fn=utf16_len):
last_msg = await _telegram_send_text_chunk(
bot, int_chat_id, chunk, send_parse_mode, _has_html, text_kwargs
)
bot, int_chat_id, chunk, send_parse_mode, _has_html, text_kwargs)
for media_path, is_voice in media_files:
if not os.path.exists(media_path):
warning = f"Media file not found, skipping: {media_path}"
logger.warning(warning)
warnings.append(warning)
# Caption mode suppressed the text send; if the file it was
# meant to caption is gone, deliver the words on their own.
# Caption mode suppressed the text send; if the file it was meant to
# caption is gone, deliver the words on their own.
if _tg_caption is not None and last_msg is None:
try:
last_msg = await _send_telegram_message_with_retry(
bot, chat_id=int_chat_id, text=_tg_caption,
parse_mode=send_parse_mode, **text_kwargs
)
parse_mode=send_parse_mode, **text_kwargs)
_tg_caption = None # delivered — don't re-caption a later file
except Exception as _cap_err:
logger.warning(
"Telegram caption-fallback send failed for missing media: %s",
_sanitize_error_text(_cap_err),
)
_sanitize_error_text(_cap_err))
continue
try:
last_msg = await _telegram_send_one_media(
bot, int_chat_id, media_path, is_voice,
caption=_tg_caption, parse_mode=send_parse_mode, has_html=_has_html,
thread_kwargs=thread_kwargs, force_document=force_document,
)
thread_kwargs=thread_kwargs, force_document=force_document)
except Exception as e:
warning = _sanitize_error_text(f"Failed to send media {media_path}: {e}")
logger.error(warning)
warnings.append(warning)
if last_msg is None:
error = "No deliverable text or media remained after processing MEDIA tags"
if warnings:
return {"error": error, "warnings": warnings}
return {"error": error}
result = {
"success": True,
"platform": "telegram",
"chat_id": chat_id,
"message_id": str(last_msg.message_id),
}
if warnings:
result["warnings"] = warnings
return result
return {"error": _NO_DELIVERABLE, **({"warnings": warnings} if warnings else {})}
return _success("telegram", chat_id, warnings, message_id=str(last_msg.message_id))
except ImportError:
return {"error": "python-telegram-bot not installed. Run: pip install python-telegram-bot"}
except Exception as e:
@@ -408,9 +359,24 @@ def _live_runner():
return None
def _live_adapter(platform, *, lookup_failed_warning=None):
"""Return ``(runner, adapter)`` for the running gateway, or ``(runner, None)``.
A runner whose adapter lookup raises is logged when a warning is given, never
silently swallowed (a silent fall-through could recreate a reconnect storm)."""
runner = _live_runner()
if runner is None:
return None, None
try:
return runner, runner.adapters.get(platform)
except Exception:
if lookup_failed_warning:
logger.warning(lookup_failed_warning, exc_info=True)
return runner, None
def _plugin_standalone_sender(platform_name, *, label=None, discover=True):
"""Return ``(standalone_sender_fn, None)`` for a registered platform plugin,
or ``(None, error_dict)``. ``discover`` runs the idempotent plugin scan first."""
"""``(standalone_sender_fn, None)`` for a registered plugin or ``(None, error_dict)``;
``discover`` runs the idempotent plugin scan first."""
from gateway.platform_registry import platform_registry
if discover:
@@ -423,8 +389,7 @@ def _plugin_standalone_sender(platform_name, *, label=None, discover=True):
async def _registry_standalone_send(platform_name, pconfig, chat_id, message, thread_id=None):
"""One-shot text send through a migrated plugin's ``standalone_sender_fn``
(the former inline ``_send_<platform>`` helpers now live in the plugins)."""
"""One-shot text send through a plugin's ``standalone_sender_fn``."""
sender, err = _plugin_standalone_sender(platform_name)
if err:
return err
@@ -432,13 +397,10 @@ async def _registry_standalone_send(platform_name, pconfig, chat_id, message, th
async def _resolve_slack_user_target(token, chat_id):
"""Resolve ``user:U...`` / ``user_name:<handle>`` to a D... DM conversation.
chat.postMessage needs a conversation ID, so user targets go through
conversations.open; ``user_name:`` first maps to a user ID via users.list
(stable handle match only). Other ids pass through unchanged.
Returns ``(chat_id, None)`` or ``(None, error_dict)``.
"""
"""Resolve ``user:U...`` / ``user_name:<handle>`` to a D... DM conversation
(chat.postMessage needs a conversation ID). ``user_name:`` maps to a user id via
users.list first (stable handle match only); other ids pass through unchanged.
Returns ``(chat_id, None)`` or ``(None, error_dict)``."""
if not (chat_id.startswith("user:") or chat_id.startswith("user_name:")):
return chat_id, None
try:
@@ -447,19 +409,17 @@ async def _resolve_slack_user_target(token, chat_id):
return None, {"error": "aiohttp not installed. Run: pip install aiohttp"}
try:
from gateway.platforms.base import resolve_proxy_url, proxy_kwargs_for_aiohttp
_proxy = resolve_proxy_url()
_sess_kw, _req_kw = proxy_kwargs_for_aiohttp(_proxy)
base_url = "https://slack.com/api"
_sess_kw, _req_kw = proxy_kwargs_for_aiohttp(resolve_proxy_url())
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
async def post_api(session, method, payload):
async with session.post(f"{base_url}/{method}", headers=headers, json=payload, **_req_kw) as resp:
async with session.post(
f"https://slack.com/api/{method}", headers=headers, json=payload, **_req_kw) as resp:
return await resp.json()
async def resolve_user_name(session, name):
query = name.strip().lstrip("@").lower()
matches = []
cursor = None
matches, cursor = [], None
for _page in range(20):
payload = {"limit": 200}
if cursor:
@@ -470,8 +430,7 @@ async def _resolve_slack_user_target(token, chat_id):
for member in data.get("members", []):
if member.get("deleted") or member.get("is_bot"):
continue
# Match the stable handle only: display/real names are
# mutable and non-unique enough to DM the wrong person.
# Stable handle only: display/real names are mutable and non-unique.
if str(member.get("name", "")).strip().lower() == query:
matches.append(member)
cursor = (data.get("response_metadata") or {}).get("next_cursor")
@@ -495,8 +454,7 @@ async def _resolve_slack_user_target(token, chat_id):
if not opened.get("ok"):
return None, _error(
f"Slack conversations.open error: {opened.get('error', 'unknown')}. "
"Check bot permissions (im:write)."
)
"Check bot permissions (im:write).")
dm_id = (opened.get("channel") or {}).get("id")
if not dm_id:
return None, _error("Slack conversations.open did not return a DM channel ID")
@@ -505,28 +463,60 @@ async def _resolve_slack_user_target(token, chat_id):
return None, _error(f"Slack DM resolution failed: {e}")
async def _send_signal(extra, chat_id, message, media_files=None):
"""Send via signal-cli JSON-RPC, text and/or attachments.
async def _signal_send_batch(post, scheduler, rl, idx, n_batches, att_batch, batch_message):
"""One Signal batch under the scheduler with rate-limit retries. None on success,
False when retries were exhausted (batch lost), error dict for a non-rate-limit RPC error."""
n = len(att_batch)
for attempt in range(1, rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS + 1):
try:
await scheduler.acquire(n)
_rpc_t0 = time.monotonic()
data = await post(att_batch, batch_message)
_rpc_duration = time.monotonic() - _rpc_t0
if "error" not in data:
await scheduler.report_rpc_duration(_rpc_duration, n)
return None
Attachments go in batches of SIGNAL_MAX_ATTACHMENTS_PER_MSG metered by the
process-wide SignalAttachmentScheduler — the same bucket the gateway
adapter uses, so tool sends and inbound replies share rate-limit state.
"""
err = data["error"]
if not rl._is_signal_rate_limit_error(err):
return _error(f"Signal RPC error on batch {idx + 1}/{n_batches}: {err}")
server_retry_after = rl._extract_retry_after_seconds(err)
scheduler.feedback(server_retry_after, n)
retry_after_label = f"{server_retry_after:.0f}s" if server_retry_after else "unknown"
if attempt >= rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS:
logger.error(
"Signal: rate-limit retries exhausted on batch %d/%d "
"(%d attachments lost, server retry_after=%s)",
idx + 1, n_batches, n, retry_after_label)
return False
logger.warning(
"Signal: rate-limited on batch %d/%d "
"(attempt %d/%d, server retry_after=%s); "
"scheduler will pace the retry",
idx + 1, n_batches, attempt, rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, retry_after_label)
except Exception as e:
if attempt >= rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS:
logger.error(
"Signal: send error on batch %d/%d after %d attempts: %s",
idx + 1, n_batches, attempt, str(e))
return False
logger.warning(
"Signal: transient error on batch %d/%d (attempt %d/%d): %s; will retry",
idx + 1, n_batches, attempt, rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, str(e))
async def _send_signal(extra, chat_id, message, media_files=None):
"""signal-cli JSON-RPC send. Attachments go in SIGNAL_MAX_ATTACHMENTS_PER_MSG batches
metered by the process-wide SignalAttachmentScheduler — the same bucket the gateway
adapter uses, so tool sends and inbound replies share rate-limit state."""
try:
import httpx
except ImportError:
return {"error": "httpx not installed"}
from gateway.platforms.signal_rate_limit import (
SIGNAL_BATCH_PACING_NOTICE_THRESHOLD,
SIGNAL_MAX_ATTACHMENTS_PER_MSG,
SIGNAL_RATE_LIMIT_MAX_ATTEMPTS,
_extract_retry_after_seconds,
_format_wait,
_is_signal_rate_limit_error,
_signal_send_timeout,
get_scheduler,
)
from gateway.platforms import signal_rate_limit as rl
from gateway.platforms.signal_format import markdown_to_signal
try:
@@ -536,129 +526,74 @@ async def _send_signal(extra, chat_id, message, media_files=None):
return {"error": "Signal account not configured"}
valid_media = media_files or []
attachment_paths = []
attachment_paths = [path for path, _is_voice in valid_media if os.path.exists(path)]
for media_path, _is_voice in valid_media:
if os.path.exists(media_path):
attachment_paths.append(media_path)
else:
if not os.path.exists(media_path):
logger.warning("Signal media file not found, skipping: %s", media_path)
# No attachments still means one (text-only) batch; with attachments
# the text rides on batch #0 so it isn't repeated per batch.
if attachment_paths:
att_batches = [
attachment_paths[i:i + SIGNAL_MAX_ATTACHMENTS_PER_MSG]
for i in range(0, len(attachment_paths), SIGNAL_MAX_ATTACHMENTS_PER_MSG)
]
else:
att_batches = [[]]
per_batch = rl.SIGNAL_MAX_ATTACHMENTS_PER_MSG
att_batches = [
attachment_paths[i:i + per_batch] for i in range(0, len(attachment_paths), per_batch)
] or [[]]
n_batches = len(att_batches)
plain_text, text_styles = markdown_to_signal(message)
def _rpc_params(text):
async def _rpc_send(text, *, id_prefix, timeout, attachments=None, styled=False):
params = {"account": account, "message": text}
if chat_id.startswith("group:"):
params["groupId"] = chat_id[6:]
else:
params["recipient"] = [chat_id]
return params
async def _rpc_send(params, *, id_prefix, timeout):
payload = {
"jsonrpc": "2.0",
"method": "send",
"params": params,
"id": f"{id_prefix}_{int(time.time() * 1000)}",
}
async with httpx.AsyncClient(timeout=timeout) as client:
return await client.post(f"{http_url}/api/v1/rpc", json=payload)
async def _post(batch_attachments, batch_message):
params = _rpc_params(batch_message)
if batch_message and text_styles:
if styled and text and text_styles:
if len(text_styles) == 1:
params["textStyle"] = text_styles[0]
else:
params["textStyles"] = text_styles
if batch_attachments:
params["attachments"] = batch_attachments
timeout = _signal_send_timeout(len(batch_attachments) if batch_attachments else 0)
resp = await _rpc_send(params, id_prefix="send", timeout=timeout)
if attachments:
params["attachments"] = attachments
payload = {
"jsonrpc": "2.0",
"method": "send",
"params": params,
"id": f"{id_prefix}_{int(time.time() * 1000)}"}
async with httpx.AsyncClient(timeout=timeout) as client:
return await client.post(f"{http_url}/api/v1/rpc", json=payload)
async def _post(batch_attachments, batch_message):
resp = await _rpc_send(
batch_message, id_prefix="send", timeout=rl._signal_send_timeout(len(batch_attachments)),
attachments=batch_attachments, styled=True)
resp.raise_for_status()
return resp.json()
async def _send_inline_notice(text: str) -> None:
"""Best-effort one-shot RPC for a user-facing pacing notice."""
try:
await _rpc_send(_rpc_params(text), id_prefix="notice", timeout=30.0)
except Exception as _e:
logger.warning("Signal: inline notice failed: %s", _e)
scheduler = get_scheduler()
scheduler = rl.get_scheduler()
logger.info(
"send_message Signal: scheduler state=%s, %d attachment(s) in %d batch(es)",
scheduler.state(), len(attachment_paths), len(att_batches),
)
scheduler.state(), len(attachment_paths), n_batches)
failed_batches: list[int] = []
for idx, att_batch in enumerate(att_batches):
n = len(att_batch)
if n > 0:
estimated = scheduler.estimate_wait(n)
if estimated >= SIGNAL_BATCH_PACING_NOTICE_THRESHOLD:
await _send_inline_notice(
f"(More images coming — pausing ~{_format_wait(estimated)} "
f"for Signal rate limit, batch {idx + 1}/{len(att_batches)}.)"
)
if estimated >= rl.SIGNAL_BATCH_PACING_NOTICE_THRESHOLD:
# Best-effort one-shot RPC for a user-facing pacing notice.
notice = (
f"(More images coming — pausing ~{rl._format_wait(estimated)} "
f"for Signal rate limit, batch {idx + 1}/{n_batches}.)")
try:
await _rpc_send(notice, id_prefix="notice", timeout=30.0)
except Exception as _e:
logger.warning("Signal: inline notice failed: %s", _e)
batch_message = plain_text if idx == 0 else ""
for attempt in range(1, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS + 1):
try:
await scheduler.acquire(n)
_rpc_t0 = time.monotonic()
data = await _post(att_batch, batch_message)
_rpc_duration = time.monotonic() - _rpc_t0
if "error" not in data:
await scheduler.report_rpc_duration(_rpc_duration, n)
break
err = data["error"]
if not _is_signal_rate_limit_error(err):
return _error(f"Signal RPC error on batch {idx + 1}/{len(att_batches)}: {err}")
server_retry_after = _extract_retry_after_seconds(err)
scheduler.feedback(server_retry_after, n)
if attempt >= SIGNAL_RATE_LIMIT_MAX_ATTEMPTS:
failed_batches.append(idx + 1)
logger.error(
"Signal: rate-limit retries exhausted on batch %d/%d "
"(%d attachments lost, server retry_after=%s)",
idx + 1, len(att_batches), n,
f"{server_retry_after:.0f}s" if server_retry_after else "unknown",
)
break
logger.warning(
"Signal: rate-limited on batch %d/%d "
"(attempt %d/%d, server retry_after=%s); "
"scheduler will pace the retry",
idx + 1, len(att_batches),
attempt, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS,
f"{server_retry_after:.0f}s" if server_retry_after else "unknown",
)
except Exception as e:
if attempt >= SIGNAL_RATE_LIMIT_MAX_ATTEMPTS:
failed_batches.append(idx + 1)
logger.error(
"Signal: send error on batch %d/%d after %d attempts: %s",
idx + 1, len(att_batches), attempt, str(e)
)
break
logger.warning(
"Signal: transient error on batch %d/%d (attempt %d/%d): %s; will retry",
idx + 1, len(att_batches), attempt, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, str(e)
)
outcome = await _signal_send_batch(
_post, scheduler, rl, idx, n_batches, att_batch, plain_text if idx == 0 else "")
if outcome is False:
failed_batches.append(idx + 1)
elif outcome is not None:
return outcome
warnings = []
if len(attachment_paths) < len(valid_media):
@@ -666,56 +601,35 @@ async def _send_signal(extra, chat_id, message, media_files=None):
if failed_batches:
warnings.append(
f"Signal rate-limited {len(failed_batches)} batch(es) "
f"(#{', #'.join(str(b) for b in failed_batches)})"
)
f"(#{', #'.join(str(b) for b in failed_batches)})")
if failed_batches and len(failed_batches) == len(att_batches):
if failed_batches and len(failed_batches) == n_batches:
return _error(
f"Signal: every batch ({len(att_batches)}) hit rate limit; "
f"no attachments delivered"
)
f"Signal: every batch ({n_batches}) hit rate limit; no attachments delivered")
result = {"success": True, "platform": "signal", "chat_id": _display_chat_id("signal", chat_id)}
if warnings:
result["warnings"] = warnings
return result
return _success("signal", _display_chat_id("signal", chat_id), warnings)
except Exception as e:
return _error(f"Signal send failed: {e}")
async def _send_matrix_via_adapter(pconfig, chat_id, message, media_files=None, thread_id=None):
"""Send via the Matrix adapter so native media uploads are preserved.
Prefer the live gateway adapter: one persistent olm/megolm session for all
sends. Ephemeral per-send connects re-init E2EE and claim one-time keys,
which under bursts exhausts recipient OTKs and silently drops messages —
so the ephemeral connect/disconnect path is only for standalone/cron.
"""
"""Matrix adapter send (native media preserved). Prefer the live gateway adapter's
persistent olm/megolm session: ephemeral per-send connects re-init E2EE and claim
one-time keys, which under bursts exhausts recipient OTKs and silently drops
messages — so the ephemeral connect/disconnect path is only for standalone/cron."""
media_files = media_files or []
metadata = {"thread_id": thread_id} if thread_id else None
# A runner that exists but whose adapter lookup fails is logged, not
# swallowed: a silent fall-through would recreate the reconnect storm.
live_adapter = None
runner = _live_runner()
if runner is not None:
try:
from gateway.config import Platform
live_adapter = runner.adapters.get(Platform.MATRIX)
except Exception:
logger.warning(
"Matrix: live gateway adapter lookup failed; falling back to an "
"ephemeral connect (may re-init E2EE per send)",
exc_info=True,
)
live_adapter = None
from gateway.config import Platform
_, live_adapter = _live_adapter(
Platform.MATRIX,
lookup_failed_warning=(
"Matrix: live gateway adapter lookup failed; falling back to an "
"ephemeral connect (may re-init E2EE per send)"))
if live_adapter is not None:
# Owned by the gateway — must NOT be disconnected; return before the
# ephemeral adapter (and its ``finally`` disconnect) exists.
return await _matrix_send_core(
live_adapter, chat_id, message, media_files, metadata
)
return await _matrix_send_core(live_adapter, chat_id, message, media_files, metadata)
# --- Fallback: ephemeral adapter (standalone / cron context) ---
try:
@@ -725,12 +639,9 @@ async def _send_matrix_via_adapter(pconfig, chat_id, message, media_files=None,
adapter = MatrixAdapter(pconfig)
try:
connected = await adapter.connect()
if not connected:
if not await adapter.connect():
return _error("Matrix connect failed")
return await _matrix_send_core(
adapter, chat_id, message, media_files, metadata
)
return await _matrix_send_core(adapter, chat_id, message, media_files, metadata)
except Exception as e:
return _error(f"Matrix send failed: {e}")
finally:
@@ -763,19 +674,12 @@ async def _matrix_send_core(adapter, chat_id, message, media_files, metadata):
else:
method = adapter.send_document
last_result = await method(chat_id, media_path, metadata=metadata)
if not last_result.success:
return _error(f"Matrix media send failed: {last_result.error}")
if last_result is None:
return {"error": "No deliverable text or media remained after processing MEDIA tags"}
return {
"success": True,
"platform": "matrix",
"chat_id": chat_id,
"message_id": last_result.message_id,
}
return {"error": _NO_DELIVERABLE}
return _success("matrix", chat_id, message_id=last_result.message_id)
async def _send_weixin(pconfig, chat_id, message, media_files=None):
@@ -789,12 +693,8 @@ async def _send_weixin(pconfig, chat_id, message, media_files=None):
try:
return await send_weixin_direct(
extra=pconfig.extra,
token=pconfig.token,
chat_id=chat_id,
message=message,
media_files=media_files,
)
extra=pconfig.extra, token=pconfig.token, chat_id=chat_id,
message=message, media_files=media_files)
except Exception as e:
return _error(f"Weixin send failed: {e}")
@@ -810,16 +710,14 @@ async def _send_bluebubbles(extra, chat_id, message):
try:
from gateway.config import PlatformConfig
pconfig = PlatformConfig(extra=extra)
adapter = BlueBubblesAdapter(pconfig)
connected = await adapter.connect()
if not connected:
adapter = BlueBubblesAdapter(PlatformConfig(extra=extra))
if not await adapter.connect():
return _error("BlueBubbles: failed to connect to server")
try:
result = await adapter.send(chat_id, message)
if not result.success:
return _error(f"BlueBubbles send failed: {result.error}")
return {"success": True, "platform": "bluebubbles", "chat_id": chat_id, "message_id": result.message_id}
return _success("bluebubbles", chat_id, message_id=result.message_id)
finally:
await adapter.disconnect()
except Exception as e:
@@ -833,8 +731,7 @@ async def _send_qqbot(pconfig, chat_id, message):
except ImportError:
return _error("QQBot direct send requires httpx. Run: pip install httpx")
# Profile-scoped secret lookup so a multiplex profile never borrows
# another profile's QQ credentials.
# Profile-scoped lookup so a multiplex profile never borrows another's QQ credentials.
from gateway.config import _getenv
extra = pconfig.extra or {}
@@ -846,36 +743,28 @@ async def _send_qqbot(pconfig, chat_id, message):
try:
async with httpx.AsyncClient(timeout=15) as client:
# Step 1: Get access token
token_resp = await client.post(
"https://bots.qq.com/app/getAppAccessToken",
json={"appId": str(appid), "clientSecret": str(secret)},
)
json={"appId": str(appid), "clientSecret": str(secret)})
if token_resp.status_code != 200:
return _error(f"QQBot token request failed: {token_resp.status_code}")
token_data = token_resp.json()
access_token = token_data.get("access_token")
access_token = token_resp.json().get("access_token")
if not access_token:
return _error("QQBot: no access_token in response")
# Step 2: QQ Bot API has separate endpoints for guild channels, C2C
# (private) and groups; try them in that order, first 2xx wins.
# Separate endpoints for guild channels, C2C (private) and groups; first 2xx wins.
headers = {
"Authorization": f"QQBot {access_token}",
"Content-Type": "application/json",
}
"Authorization": f"QQBot {access_token}", "Content-Type": "application/json"}
payload = {"content": message[:4000], "msg_type": 0}
endpoints = (
("channel", f"https://api.sgroup.qq.com/channels/{chat_id}/messages"),
("c2c", f"https://api.sgroup.qq.com/v2/users/{chat_id}/messages"),
("group", f"https://api.sgroup.qq.com/v2/groups/{chat_id}/messages"),
)
("group", f"https://api.sgroup.qq.com/v2/groups/{chat_id}/messages"))
statuses = []
for kind, url in endpoints:
resp = await client.post(url, json=payload, headers=headers)
if resp.status_code in {200, 201}:
return {"success": True, "platform": "qqbot", "chat_id": chat_id,
"message_id": resp.json().get("id")}
return _success("qqbot", chat_id, message_id=resp.json().get("id"))
statuses.append(f"{kind}={resp.status_code}")
return _error(f"QQBot send failed: {' '.join(statuses)}")
except Exception as e:
@@ -883,8 +772,8 @@ async def _send_qqbot(pconfig, chat_id, message):
async def _send_yuanbao(chat_id, message, media_files=None):
"""Send via the running Yuanbao adapter's persistent WebSocket (no
throwaway client possible). chat_id: ``group:<code>``, ``direct:<id>`` or ``<id>``."""
"""Send via the running Yuanbao adapter's persistent WebSocket (no throwaway client
possible). chat_id: ``group:<code>``, ``direct:<id>`` or ``<id>``."""
try:
from gateway.platforms.yuanbao import get_active_adapter, send_yuanbao_direct
except ImportError:
@@ -894,8 +783,7 @@ async def _send_yuanbao(chat_id, message, media_files=None):
if adapter is None:
return _error(
"Yuanbao adapter is not running. "
"Start the gateway with yuanbao platform enabled first."
)
"Start the gateway with yuanbao platform enabled first.")
try:
return await send_yuanbao_direct(adapter, chat_id, message, media_files=media_files)
tools/send_message_targets.py
+77 -127
View File
@@ -7,12 +7,10 @@ logger = logging.getLogger("tools.send_message_tool")
_TELEGRAM_TOPIC_TARGET_RE = re.compile(r"^\s*(-?\d+)(?::(\d+))?\s*$")
_FEISHU_TARGET_RE = re.compile(r"^\s*((?:oc|ou|on|chat|open)_[-A-Za-z0-9]+)(?::([-A-Za-z0-9_]+))?\s*$")
# Slack conversation IDs: C (public channel), G (private/group channel), D (DM).
# Must be uppercase alphanumeric, 9+ chars. User IDs (U...) are parsed as
# explicit user targets (``user:U...``) and are converted to D... conversations
# via conversations.open before chat.postMessage — posting directly to a U/W
# ID fails because the API requires a conversation ID. ``@handle`` targets are
# resolved through users.list first (``user_name:...``).
# Slack conversation IDs: C (public), G (private/group), D (DM); uppercase alnum, 9+ chars.
# User IDs (U...) become ``user:U...`` and are opened as D... conversations before
# chat.postMessage (posting straight to a U/W id fails); ``@handle`` -> ``user_name:...``
# is resolved through users.list first.
_SLACK_TARGET_RE = re.compile(r"^\s*([CGD][A-Z0-9]{8,})\s*$")
_SLACK_USER_ID_RE = re.compile(r"^\s*(U[A-Z0-9]{8,})\s*$")
_SLACK_USER_NAME_RE = re.compile(r"^\s*@([A-Za-z0-9._-]{1,80})\s*$")
@@ -23,47 +21,32 @@ _WEIXIN_TARGET_RE = re.compile(r"^\s*((?:wxid|gh|v\d+|wm|wb)_[A-Za-z0-9_-]+|[A-Z
_YUANBAO_TARGET_RE = re.compile(r"^\s*((?:group|direct):[^:]+)\s*$")
# Discord snowflake IDs are numeric, same regex pattern as Telegram topic targets.
_NUMERIC_TOPIC_RE = _TELEGRAM_TOPIC_TARGET_RE
# Platforms that address recipients by phone number and accept E.164 format
# (with a leading '+'). Without this, "+15551234567" fails the isdigit() check
# below and falls through to channel-name resolution, which has no way to
# resolve a raw phone number. Keeping the '+' preserves the E.164 form that
# downstream adapters (signal, etc.) expect.
# Platforms addressing recipients by E.164 phone number ("+1555..."): without this the
# '+' fails the isdigit() rule and falls through to channel-name resolution, which cannot
# resolve a raw number. The '+' is kept because the adapters expect the E.164 form.
_PHONE_PLATFORMS = frozenset({"photon", "signal", "sms", "whatsapp"})
_E164_TARGET_RE = re.compile(r"^\s*\+(\d{7,15})\s*$")
# Photon DM chat GUID (mirrors _DM_CHAT_GUID_RE in the photon adapter).
_PHOTON_DM_GUID_RE = re.compile(r"^any;-;\+\d{6,}$")
# WhatsApp JIDs: group chats (<digits>@g.us), individual users
# (<phone>@s.whatsapp.net), linked identities (<id>@lid), and broadcast /
# newsletter chats. These are explicit native targets the bridge accepts
# verbatim — they must NOT fall through to home-channel resolution.
# WhatsApp JIDs (groups <digits>@g.us, users <phone>@s.whatsapp.net, linked ids @lid,
# broadcast/newsletter): native targets the bridge accepts verbatim — never home-channel.
_WHATSAPP_JID_RE = re.compile(
r"^\s*[\w-]+@(?:g\.us|s\.whatsapp\.net|lid|broadcast|newsletter)\s*$",
re.IGNORECASE,
)
# Buzz channels and DMs use native UUID identifiers. They are explicit
# targets and must never substitute the configured home channel.
r"^\s*[\w-]+@(?:g\.us|s\.whatsapp\.net|lid|broadcast|newsletter)\s*$", re.IGNORECASE)
# Buzz channels/DMs are native UUIDs: explicit targets, never the home channel.
_BUZZ_UUID_RE = re.compile(
r"^\s*[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\s*$",
re.IGNORECASE,
)
# Email addresses — a valid email like "user@domain.com" should be treated as
# an explicit target for the email platform, not fall through to channel-name
# resolution which has no way to resolve a raw address.
r"^\s*[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\s*$", re.IGNORECASE)
# A valid address is an explicit email target, not a channel name to resolve.
_EMAIL_TARGET_RE = re.compile(r"^\s*[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\s*$")
# Most platforms read their home channel from "<PLATFORM>_HOME_CHANNEL", but a
# few diverge. Email reads EMAIL_HOME_ADDRESS (see gateway/config.py), so the
# generic "<PLATFORM>_HOME_CHANNEL" hint would point users at a variable that is
# never read. Map the exceptions so the error guidance is actually actionable.
# Home-channel env var exceptions to "<PLATFORM>_HOME_CHANNEL" (email reads
# EMAIL_HOME_ADDRESS, see gateway/config.py) so the error guidance is actionable.
_HOME_CHANNEL_ENV_OVERRIDES = {"email": "EMAIL_HOME_ADDRESS"}
_UNRESOLVED = object() # sentinel: stop parsing, target is NOT explicit (skip generic rules)
# Per-platform explicit-target parsers: target_ref -> (chat_id, thread_id),
# None to fall through to the generic rules in _parse_target_ref, or
# _UNRESOLVED. Order inside each parser matters (e.g. Slack thread form
# before bare conversation id).
# Per-platform explicit-target parsers: target_ref -> (chat_id, thread_id), None to fall
# through to the generic rules in _parse_target_ref, or _UNRESOLVED. Order inside each
# parser matters (e.g. Slack thread form before bare conversation id).
def _parse_regex_groups(regex, *, thread_group=True):
"""Explicit when ``regex`` fully matches: chat_id = group 1, thread = group 2 (or None)."""
def parse(ref):
@@ -74,6 +57,13 @@ def _parse_regex_groups(regex, *, thread_group=True):
return parse
def _parse_regex_stripped(regex):
"""Explicit when ``regex`` fully matches; returns the stripped ref verbatim."""
def parse(ref):
return (ref.strip(), None) if regex.fullmatch(ref) else None
return parse
def _parse_telegram(ref):
# "<chat_id>[:<topic_id>]" or an @username (usernames must not be force-int'd).
parsed = _parse_regex_groups(_TELEGRAM_TOPIC_TARGET_RE)(ref)
@@ -85,24 +75,26 @@ def _parse_telegram(ref):
return (username, None) if username else None
# (regex, chat_id template, thread comes from group 2) — thread form before bare id.
_SLACK_FORMS = (
(_SLACK_THREAD_TARGET_RE, "{}", True),
(_SLACK_TARGET_RE, "{}", False),
(_SLACK_USER_ID_RE, "user:{}", False),
(_SLACK_MENTION_RE, "user:{}", False),
(_SLACK_USER_NAME_RE, "user_name:{}", False))
def _parse_slack(ref):
match = _SLACK_THREAD_TARGET_RE.fullmatch(ref)
if match:
return match.group(1), match.group(2)
match = _SLACK_TARGET_RE.fullmatch(ref)
if match:
return match.group(1), None
match = _SLACK_USER_ID_RE.fullmatch(ref) or _SLACK_MENTION_RE.fullmatch(ref)
if match:
return f"user:{match.group(1)}", None
match = _SLACK_USER_NAME_RE.fullmatch(ref)
return (f"user_name:{match.group(1)}", None) if match else None
for regex, template, has_thread in _SLACK_FORMS:
match = regex.fullmatch(ref)
if match:
return template.format(match.group(1)), (match.group(2) if has_thread else None)
return None
def _parse_matrix(ref):
# "<room>:$<event_id>" addresses a thread (rfind: room ids contain ':').
# Bare "!room" / "@user" are explicit too, but via the generic rule so the
# numeric check keeps precedence exactly as before.
# "<room>:$<event_id>" addresses a thread (rfind: room ids contain ':'). Bare "!room" /
# "@user" are explicit too, but via the generic rule so the numeric check keeps precedence.
trimmed = ref.strip()
split_idx = trimmed.rfind(":$")
if split_idx > 0:
@@ -110,16 +102,9 @@ def _parse_matrix(ref):
return None
def _parse_regex_stripped(regex):
"""Explicit when ``regex`` fully matches; returns the stripped ref verbatim."""
def parse(ref):
return (ref.strip(), None) if regex.fullmatch(ref) else None
return parse
def _parse_yuanbao(ref):
# "group:<code>" / "direct:<id>"; a bare number is a group code, never a
# generic numeric chat id (yuanbao never falls through to the generic rules).
# "group:<code>" / "direct:<id>"; a bare number is a group code, never a generic
# numeric chat id (yuanbao never falls through to the generic rules).
match = _YUANBAO_TARGET_RE.fullmatch(ref)
if match:
return match.group(1), None
@@ -129,8 +114,8 @@ def _parse_yuanbao(ref):
def _parse_nonempty(ref):
# ntfy topics and WeCom ids (wr/wc groups, wo/bare users — the adapter
# picks the send command) are explicit whenever non-empty.
# ntfy topics and WeCom ids (wr/wc groups, wo/bare users — the adapter picks the
# send command) are explicit whenever non-empty.
stripped = ref.strip()
return (stripped, None) if stripped else None
@@ -160,18 +145,15 @@ _PLATFORM_PARSERS = {
"signal": _parse_signal,
"wecom": _parse_nonempty,
# Photon DM GUIDs are adapter-native ids (mirrors the react handler).
"photon": _parse_regex_stripped(_PHOTON_DM_GUID_RE),
}
"photon": _parse_regex_stripped(_PHOTON_DM_GUID_RE)}
def _parse_target_ref(platform_name: str, target_ref: str):
"""Parse a tool target into (chat_id, thread_id, explicit).
Platform parser first, then the shared rules in this order: E.164 phone
numbers (keeping the '+' the signal/sms/whatsapp adapters expect), bare
numeric ids, Matrix ``!room``/``@user``, XMPP JIDs (user@server or
room@conference.server). Anything else is not explicit and goes to
channel-directory resolution.
Platform parser first, then the shared rules in this order: E.164 phone numbers
(keeping the '+' the signal/sms/whatsapp adapters expect), bare numeric ids, Matrix
``!room``/``@user``, XMPP JIDs. Anything else goes to channel-directory resolution.
"""
parser = _PLATFORM_PARSERS.get(platform_name)
if parser is not None:
@@ -196,22 +178,18 @@ def resolve_send_target(
) -> tuple[str | None, str | None, str | None]:
"""Resolve one send target the same way for every caller (model tool, CLI, cron).
Channel-directory IDs are trusted. Plugin platforms must explicitly parse
native target syntax; for the model-facing send tool (the default), a
target that can't be resolved is an error — the model can read the error
and pick a listed target instead.
Channel-directory IDs are trusted. Plugin platforms must explicitly parse native
target syntax; for the model-facing send tool (the default) an unresolvable target
is an error the model can read and pick a listed target instead.
``pass_unresolved_references=True`` restores the old pass-through behavior for
callers that have no model in the loop (cron delivering a stored job's
output, react/unreact on platform-native message ids): if the target
can't be resolved and the platform is built in, or is a plugin platform
that declares no parser, the string is handed to the adapter exactly as
written and the adapter decides whether it's valid. A plugin platform
that DOES declare a parser stays strict for every caller — its parser is
the authority on native syntax.
``pass_unresolved_references=True`` is for callers with no model in the loop (cron
delivering a stored job's output, react/unreact on platform-native message ids): an
unresolvable target on a built-in platform, or on a plugin platform that declares no
parser, is handed to the adapter exactly as written and the adapter decides. A plugin
platform that DOES declare a parser stays strict for every caller.
The optional validator has the final say over parser-normalized,
directory-resolved, and passed-through IDs alike.
The optional validator has the final say over parser-normalized, directory-resolved,
and passed-through IDs alike.
"""
from gateway.config import Platform
from gateway.platform_registry import platform_registry
@@ -224,9 +202,7 @@ def resolve_send_target(
try:
verdict = entry.validate_target_ref_fn(candidate)
except Exception:
logger.debug(
"Plugin target validator failed for %s", platform_name, exc_info=True
)
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
@@ -234,13 +210,15 @@ def resolve_send_target(
return f"Invalid target '{target_ref}' on {platform_name}: {verdict}"
return f"Invalid target '{target_ref}' on {platform_name}"
def _validated(chat_id, thread_id):
error = _validate(chat_id)
return (None, None, error) if error else (chat_id, thread_id, None)
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
)
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 (
@@ -248,31 +226,15 @@ def resolve_send_target(
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))
):
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,
)
None, None,
f"Target parser for platform '{platform_name}' returned an invalid result")
return _validated(*parsed)
parsed_chat_id, parsed_thread_id, explicit = _parse_target_ref(
platform_name, target_ref
)
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,
)
return _validated(parsed_chat_id, parsed_thread_id)
resolution_failed = False
try:
@@ -283,16 +245,8 @@ def resolve_send_target(
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,
)
parsed_chat_id, parsed_thread_id, _ = _parse_target_ref(platform_name, resolved)
return _validated(parsed_chat_id or resolved, parsed_thread_id)
is_builtin = platform_name in {member.value for member in Platform}
if entry is None and not is_builtin:
@@ -306,24 +260,20 @@ def resolve_send_target(
logger.debug(
"Handing unresolved target '%s' to the %s adapter unchanged "
"(the adapter validates it)",
target_ref, platform_name,
)
target_ref, platform_name)
return target_ref, None, None
if entry is not None and entry.source == "plugin" and not is_builtin:
if pass_unresolved_references and entry.parse_target_ref_fn is None:
return _pass_through_unresolved()
return (
None,
None,
None, None,
f"Could not resolve '{target_ref}' on {platform_name}. "
"The plugin parser did not recognize it and no channel-directory entry matched.",
)
"The plugin parser did not recognize it and no channel-directory entry matched.")
if pass_unresolved_references:
return _pass_through_unresolved()
hint = (
"Try using a numeric channel ID instead."
if resolution_failed
else "Use send_message(action='list') to see available targets."
)
else "Use send_message(action='list') to see available targets.")
return None, None, f"Could not resolve '{target_ref}' on {platform_name}. {hint}"
tools/send_message_tool.py
+201 -382
View File
@@ -1,7 +1,7 @@
"""Send Message Tool -- cross-channel messaging via platform APIs.
Sends a message to a user or channel on any connected messaging platform
(Telegram, Discord, Slack). Supports listing available targets and resolving
(Telegram, Discord, Slack, ...), lists available targets and resolves
human-friendly channel names to IDs. Works in both CLI and gateway contexts.
"""
@@ -14,58 +14,22 @@ from agent.secret_scope import get_secret
logger = logging.getLogger(__name__)
# Re-exported: tests and sibling modules import these via tools.send_message_tool.
from tools.send_message_targets import ( # noqa: F401
_BUZZ_UUID_RE,
_E164_TARGET_RE,
_EMAIL_TARGET_RE,
_FEISHU_TARGET_RE,
_HOME_CHANNEL_ENV_OVERRIDES,
_NUMERIC_TOPIC_RE,
_PHONE_PLATFORMS,
_PHOTON_DM_GUID_RE,
_SLACK_MENTION_RE,
_SLACK_TARGET_RE,
_SLACK_THREAD_TARGET_RE,
_SLACK_USER_ID_RE,
_SLACK_USER_NAME_RE,
_TELEGRAM_TOPIC_TARGET_RE,
_WEIXIN_TARGET_RE,
_WHATSAPP_JID_RE,
_YUANBAO_TARGET_RE,
_parse_target_ref,
resolve_send_target,
)
_HOME_CHANNEL_ENV_OVERRIDES, _SLACK_USER_ID_RE, _parse_target_ref, resolve_send_target)
from tools.send_message_senders import ( # noqa: F401
_AUDIO_EXTS,
_CAPTIONABLE_EXTS,
_DEFAULT_CAPTION_LIMIT,
_GENERIC_SECRET_ASSIGN_RE,
_IMAGE_EXTS,
_TELEGRAM_CAPTION_LIMIT,
_TELEGRAM_SEND_AUDIO_EXTS,
_URL_SECRET_QUERY_RE,
_VIDEO_EXTS,
_VOICE_EXTS,
_display_chat_id,
_error,
_is_telegram_thread_not_found,
_live_runner,
_matrix_send_core,
_media_caption_split,
_plugin_standalone_sender,
_registry_standalone_send,
_resolve_slack_user_target,
_sanitize_error_text,
_send_bluebubbles,
_send_matrix_via_adapter,
_send_qqbot,
_send_signal,
_send_telegram,
_send_telegram_message_with_retry,
_send_weixin,
_send_yuanbao,
_telegram_retry_delay,
)
_AUDIO_EXTS, _DEFAULT_CAPTION_LIMIT, _IMAGE_EXTS, _TELEGRAM_CAPTION_LIMIT, _VIDEO_EXTS,
_VOICE_EXTS, _error, _live_adapter, _media_caption_split, _plugin_standalone_sender,
_registry_standalone_send, _resolve_slack_user_target, _sanitize_error_text,
_send_bluebubbles, _send_matrix_via_adapter, _send_qqbot, _send_signal, _send_telegram,
_send_weixin, _send_yuanbao)
from tools.registry import tool_error
# NOTE: ``send_message`` is intentionally NOT registered as an agent-callable model
# tool (the agent must not fire cross-platform messages on its own). The send engine
# here is the shared transport for cron delivery, the ``hermes send`` CLI, the gateway
# kanban notifier and the opt-in MCP server, which import the helpers directly.
def prepare_send_message_platforms() -> None:
"""Load enabled standalone plugins before tool schemas/cache keys are built."""
@@ -90,10 +54,16 @@ def _split_target(target: str):
return parts[0].strip().lower(), (parts[1].strip() if len(parts) > 1 else None)
def _live_adapter(platform):
"""Return the running gateway's adapter for ``platform``, or None."""
runner = _live_runner()
return runner.adapters.get(platform) if runner is not None else None
def _resolve_tool_target(target: str, *, pass_unresolved_references: bool = False):
"""Split and resolve a tool ``target``; returns ``(platform_name, chat_id, thread_id, error)``.
``chat_id`` is None when no ref was given (caller falls back to the home channel)."""
platform_name, target_ref = _split_target(target)
prepare_send_message_platforms()
if not target_ref:
return platform_name, None, None, None
chat_id, thread_id, resolution_error = resolve_send_target(
platform_name, target_ref, pass_unresolved_references=pass_unresolved_references)
return platform_name, chat_id, thread_id, resolution_error
def _handle_list():
@@ -106,32 +76,23 @@ def _handle_list():
def _handle_react(args, remove=False):
"""Attach (or with ``remove=True`` retract) an emoji reaction via a live
gateway adapter exposing ``add_reaction(chat_id, emoji, message_id)`` /
``remove_reaction(chat_id, message_id)`` (e.g. photon/iMessage tapbacks).
No standalone fallback: reacting needs the adapter's live message-id state.
"""
"""Attach (or with ``remove=True`` retract) an emoji reaction via a live gateway
adapter exposing ``add_reaction(chat_id, emoji, message_id)`` /
``remove_reaction(chat_id, message_id)`` (e.g. photon/iMessage tapbacks). No
standalone fallback: reacting needs the adapter's live message-id state."""
target = args.get("target", "")
emoji = (args.get("emoji") or "").strip()
message_id = (args.get("message_id") or "").strip() or None
if not target or (not remove and not emoji):
return tool_error(
"Both 'target' and 'emoji' are required when action='react'"
if not remove
else "'target' is required when action='unreact'"
)
return tool_error("'target' is required when action='unreact'" if remove
else "Both 'target' and 'emoji' are required when action='react'")
platform_name, target_ref = _split_target(target)
chat_id = None
prepare_send_message_platforms()
if target_ref:
# Platform-native ids (e.g. photon GUIDs) match no parser/directory
# entry; hand them to the adapter unchanged and let it validate.
chat_id, _thread_id, resolution_error = resolve_send_target(
platform_name, target_ref, pass_unresolved_references=True
)
if resolution_error:
return tool_error(resolution_error)
# Platform-native ids (e.g. photon GUIDs) match no parser/directory entry; hand
# them to the adapter unchanged and let it validate.
platform_name, chat_id, _thread_id, resolution_error = _resolve_tool_target(
target, pass_unresolved_references=True)
if resolution_error:
return tool_error(resolution_error)
try:
from gateway.config import Platform, load_gateway_config
@@ -147,21 +108,17 @@ def _handle_react(args, remove=False):
if not home:
return tool_error(
f"No chat specified and no home channel set for {platform_name}. "
f"Use '{platform_name}:chat_id'."
)
f"Use '{platform_name}:chat_id'.")
chat_id = home.chat_id
adapter = _live_adapter(platform)
_, adapter = _live_adapter(platform)
if adapter is None:
return tool_error(
f"Reactions require a live {platform_name} adapter in the running "
"gateway (not available from cron/standalone contexts)."
)
"gateway (not available from cron/standalone contexts).")
react_fn = getattr(adapter, "remove_reaction" if remove else "add_reaction", None)
if not callable(react_fn):
return tool_error(
f"Platform '{platform_name}' does not support message reactions."
)
return tool_error(f"Platform '{platform_name}' does not support message reactions.")
kwargs = {"chat_id": chat_id, "message_id": message_id}
if not remove:
@@ -171,9 +128,7 @@ def _handle_react(args, remove=False):
result = _run_async(react_fn(**kwargs))
except Exception as e:
return json.dumps(_error(f"Reaction failed: {e}"))
if isinstance(result, dict):
return json.dumps(result)
return json.dumps({"success": bool(result)})
return json.dumps(result if isinstance(result, dict) else {"success": bool(result)})
def _handle_send(args):
@@ -183,17 +138,9 @@ def _handle_send(args):
if not target or not message:
return tool_error("Both 'target' and 'message' are required when action='send'")
platform_name, target_ref = _split_target(target)
chat_id = None
thread_id = None
prepare_send_message_platforms()
if target_ref:
chat_id, thread_id, resolution_error = resolve_send_target(
platform_name, target_ref
)
if resolution_error:
return tool_error(resolution_error)
platform_name, chat_id, thread_id, resolution_error = _resolve_tool_target(target)
if resolution_error:
return tool_error(resolution_error)
from tools.interrupt import is_interrupted
if is_interrupted():
@@ -211,9 +158,9 @@ def _handle_send(args):
from gateway.platforms.base import BasePlatformAdapter
# Capture [[as_document]] before extract_media strips it: image files then
# go through send_document so the original bytes survive (Telegram's
# sendPhoto recompresses to 1280px).
# Capture [[as_document]] before extract_media strips it: image files then go
# through send_document so the original bytes survive (Telegram's sendPhoto
# recompresses to 1280px).
force_document_attachments = "[[as_document]]" in message
media_files, cleaned_message = BasePlatformAdapter.extract_media(message)
@@ -237,18 +184,13 @@ 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,
}
send_kwargs = {"thread_id": thread_id, "media_files": media_files,
"force_document": force_document_attachments}
# Only custom plugin handlers receive the complete typed request; the
# built-in call contract stays exact.
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, **send_kwargs)
)
result = _run_async(_send_to_platform(platform, pconfig, chat_id, cleaned_message, **send_kwargs))
if isinstance(result, dict) and result.get("success"):
if used_home_channel:
result["note"] = f"Sent to {platform_name} home channel (chat_id: {chat_id})"
@@ -265,9 +207,9 @@ def _handle_send(args):
def _resolve_platform_config(platform_name, config):
"""Return ``(platform, pconfig, registry_entry, error)`` for a send.
Plugin platforms must be registered; disabled/missing platforms error,
except Weixin, which may be configured purely via .env (synthesized
pconfig so send_message and cron delivery work without a gateway.yaml entry).
Plugin platforms must be registered; disabled/missing platforms error, except
Weixin, which may be configured purely via .env (synthesized pconfig so
send_message and cron delivery work without a gateway.yaml entry).
"""
from gateway.config import Platform
from gateway.platform_registry import platform_registry
@@ -287,8 +229,7 @@ def _resolve_platform_config(platform_name, config):
if pconfig is None:
return None, None, None, (
f"Platform '{platform_name}' is not configured. Set up credentials in "
"~/.hermes/config.yaml or environment variables."
)
"~/.hermes/config.yaml or environment variables.")
return platform, pconfig, entry, None
@@ -303,21 +244,17 @@ def _home_chat_id(config, platform, platform_name):
home = HomeChannel(platform=platform, chat_id=wx_home, name="Weixin Home")
if home:
return home.chat_id, None
home_env = _HOME_CHANNEL_ENV_OVERRIDES.get(
platform_name, f"{platform_name.upper()}_HOME_CHANNEL"
)
home_env = _HOME_CHANNEL_ENV_OVERRIDES.get(platform_name, f"{platform_name.upper()}_HOME_CHANNEL")
return None, (
f"No home channel set for {platform_name} to determine where to send the message. "
f"Either specify a channel directly with '{platform_name}:CHANNEL_NAME', "
f"or set a home channel via: hermes config set {home_env} <channel_id>"
)
f"or set a home channel via: hermes config set {home_env} <channel_id>")
def _slack_dm_chat_id(pconfig, chat_id):
"""Open Slack user targets as DM conversations: ``user:U...`` /
``user_name:@handle`` from the parser, or a bare U... id from session
metadata / home-channel config. chat.postMessage needs a conversation ID.
Returns ``(chat_id, None)`` or ``(None, error_dict)``."""
"""Open Slack user targets as DM conversations: ``user:U...`` / ``user_name:@handle``
from the parser, or a bare U... id from session metadata / home-channel config.
chat.postMessage needs a conversation ID. Returns ``(chat_id, None)`` or ``(None, error_dict)``."""
dm_target = chat_id
if dm_target.startswith("U") and _SLACK_USER_ID_RE.fullmatch(dm_target):
dm_target = f"user:{dm_target}"
@@ -333,13 +270,10 @@ def _mirror_sent_message(platform_name, chat_id, mirror_text, thread_id):
from gateway.mirror import mirror_to_session
from gateway.session_context import get_session_env
return bool(mirror_to_session(
platform_name,
chat_id,
mirror_text,
platform_name, chat_id, mirror_text,
source_label=get_session_env("HERMES_SESSION_PLATFORM", "cli"),
thread_id=thread_id,
user_id=get_session_env("HERMES_SESSION_USER_ID", "") or None,
))
user_id=get_session_env("HERMES_SESSION_USER_ID", "") or None))
except Exception:
return False
@@ -351,15 +285,10 @@ def _weixin_env_pconfig():
if not (wx_token and wx_account):
return None
from gateway.config import PlatformConfig
return PlatformConfig(
enabled=True,
token=wx_token,
extra={
"account_id": wx_account,
"base_url": get_secret("WEIXIN_BASE_URL", "").strip(),
"cdn_base_url": get_secret("WEIXIN_CDN_BASE_URL", "").strip(),
},
)
return PlatformConfig(enabled=True, token=wx_token, extra={
"account_id": wx_account,
"base_url": get_secret("WEIXIN_BASE_URL", "").strip(),
"cdn_base_url": get_secret("WEIXIN_CDN_BASE_URL", "").strip()})
def _describe_media_for_mirror(media_files):
@@ -378,165 +307,110 @@ def _describe_media_for_mirror(media_files):
return "[Sent document attachment]"
def _get_cron_auto_delivery_target():
"""Return the cron scheduler's auto-delivery target for the current run, if any."""
from gateway.session_context import get_session_env
platform = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM", "").strip().lower()
chat_id = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID", "").strip()
if not platform or not chat_id:
return None
thread_id = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None
return {"platform": platform, "chat_id": chat_id, "thread_id": thread_id}
def _maybe_skip_cron_duplicate_send(platform_name: str, chat_id: str, thread_id: str | None):
"""Skip redundant cron send_message calls when the scheduler will auto-deliver there."""
auto_target = _get_cron_auto_delivery_target()
if not auto_target or not (
auto_target["platform"] == platform_name
and str(auto_target["chat_id"]) == str(chat_id)
and auto_target.get("thread_id") == thread_id
):
from gateway.session_context import get_session_env
auto_platform = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM", "").strip().lower()
auto_chat_id = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID", "").strip()
if not auto_platform or not auto_chat_id:
return None
auto_thread_id = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None
if not (auto_platform == platform_name and auto_chat_id == str(chat_id) and auto_thread_id == thread_id):
return None
target_label = f"{platform_name}:{chat_id}"
if thread_id is not None:
target_label += f":{thread_id}"
return {
"success": True,
"skipped": True,
"reason": "cron_auto_delivery_duplicate_target",
"success": True, "skipped": True, "reason": "cron_auto_delivery_duplicate_target",
"target": target_label,
"note": (
f"Skipped send_message to {target_label}. This cron job will already auto-deliver "
"its final response to that same target. Put the intended user-facing content in "
"your final response instead, or use a different target if you want an additional message."
),
}
)}
def _bounded_send_error(detail, max_chars=900):
"""Bound untrusted adapter/plugin error detail returned by send_message."""
text = str(detail or "send failed")
if len(text) <= max_chars:
return text
return f"{text[: max_chars - 3]}..."
return text if len(text) <= max_chars else f"{text[: max_chars - 3]}..."
def _live_media_method(ext, is_voice, force_document):
"""Adapter media method name and kind for one live-adapter media file."""
if force_document:
return "send_document", "document"
if ext in _IMAGE_EXTS:
return "send_image_file", "image"
if ext in _VIDEO_EXTS:
return "send_video", "video"
if is_voice or ext in _AUDIO_EXTS:
return "send_voice", "audio"
return "send_document", "document"
async def _send_live_adapter_media(
adapter,
chat_id,
message,
media_files,
*,
thread_id=None,
metadata=None,
force_document=False,
):
adapter, chat_id, message, media_files, *, thread_id=None, metadata=None, force_document=False):
"""Deliver text and every media descriptor through adapter media APIs.
Adapters that only inherit the BasePlatformAdapter stub for a media kind
are treated as unsupported rather than silently no-op'd."""
caption, separate_text = _media_caption_split(
message, media_files, max_caption_len=_DEFAULT_CAPTION_LIMIT
)
message, media_files, max_caption_len=_DEFAULT_CAPTION_LIMIT)
last_result = None
if separate_text and separate_text.strip():
last_result = await adapter.send(
chat_id=chat_id, content=separate_text, metadata=metadata
)
last_result = await adapter.send(chat_id=chat_id, content=separate_text, metadata=metadata)
if not last_result.success:
return {"error": f"Adapter send failed: {_bounded_send_error(last_result.error)}"}
from gateway.platforms.base import BasePlatformAdapter
total = len(media_files)
for index, descriptor in enumerate(media_files):
invalid = {"error": f"Adapter media send failed: invalid media descriptor {index + 1}/{total}"}
if not isinstance(descriptor, (list, tuple)) or not descriptor:
return {"error": f"Adapter media send failed: invalid media descriptor {index + 1}/{total}"}
return invalid
media_path = descriptor[0]
is_voice = bool(descriptor[1]) if len(descriptor) > 1 else False
if not isinstance(media_path, str) or not media_path:
return {"error": f"Adapter media send failed: invalid media descriptor {index + 1}/{total}"}
return invalid
if not os.path.exists(media_path):
return {"error": f"Adapter media send failed: media file {index + 1}/{total} was not found"}
ext = os.path.splitext(media_path)[1].lower()
kwargs = {
"caption": caption if index == 0 else None,
"reply_to": thread_id,
"metadata": metadata,
}
if force_document:
method_name, media_kind = "send_document", "document"
elif ext in _IMAGE_EXTS:
method_name, media_kind = "send_image_file", "image"
elif ext in _VIDEO_EXTS:
method_name, media_kind = "send_video", "video"
elif is_voice or ext in _AUDIO_EXTS:
method_name, media_kind = "send_voice", "audio"
else:
method_name, media_kind = "send_document", "document"
from gateway.platforms.base import BasePlatformAdapter
method_name, media_kind = _live_media_method(ext, is_voice, force_document)
adapter_method = getattr(type(adapter), method_name, None)
base_fallback = getattr(BasePlatformAdapter, method_name)
if adapter_method is None or adapter_method is base_fallback:
if adapter_method is None or adapter_method is getattr(BasePlatformAdapter, method_name):
return {
"error": (
f"Live adapter does not implement native {media_kind} delivery; "
f"media file {index + 1}/{total} was not sent"
)
}
f"media file {index + 1}/{total} was not sent")}
kwargs = {"caption": caption if index == 0 else None, "reply_to": thread_id, "metadata": metadata}
try:
last_result = await getattr(adapter, method_name)(chat_id, media_path, **kwargs)
except asyncio.CancelledError:
raise
except Exception as exc:
return {
"error": (
f"Adapter media send failed after {index}/{total} files: "
f"{_bounded_send_error(exc)}"
)
}
if not last_result.success:
detail = _bounded_send_error(exc)
else:
if last_result.success:
continue
detail = _bounded_send_error(last_result.error or "media send failed")
return {
"error": f"Adapter media send failed after {index}/{total} files: {detail}"
}
return {"error": f"Adapter media send failed after {index}/{total} files: {detail}"}
if last_result is None:
return {"error": "No deliverable text or media remained after processing MEDIA tags"}
return {
"success": True,
"message_id": last_result.message_id,
"media_delivered": True,
}
return {"success": True, "message_id": last_result.message_id, "media_delivered": True}
async def _send_via_adapter(
platform,
pconfig,
chat_id,
chunk,
*,
thread_id=None,
media_files=None,
force_document=False,
):
"""Send via the live in-process gateway adapter (``_gateway_runner_ref``),
else the plugin's ``standalone_sender_fn`` (gateway not in this process,
e.g. cron — the runner weakref is None), else a descriptive error naming
both options. Media descriptors go through the adapter's native media
APIs under the same cross-loop rules as text.
"""
platform, pconfig, chat_id, chunk, *, thread_id=None, media_files=None, force_document=False):
"""Send via the live in-process gateway adapter (``_gateway_runner_ref``), else the
plugin's ``standalone_sender_fn`` (gateway not in this process, e.g. cron — the runner
weakref is None), else a descriptive error naming both options. Media descriptors go
through the adapter's native media APIs under the same cross-loop rules as text."""
platform_name = platform.value if hasattr(platform, "value") else str(platform)
runner = _live_runner()
adapter = None
if runner is not None:
try:
adapter = runner.adapters.get(platform)
except Exception:
adapter = None
runner, adapter = _live_adapter(platform)
if adapter is not None:
try:
metadata = {}
@@ -545,9 +419,9 @@ async def _send_via_adapter(
if platform_name == "ntfy" and chat_id:
metadata["publish_topic"] = chat_id
metadata = metadata or None
# adapter.send() uses queues/tasks bound to the gateway's loop.
# Awaiting it from another loop (the tool worker thread) deadlocks
# on a cross-loop Future, so dispatch onto the gateway loop instead.
# adapter.send() uses queues/tasks bound to the gateway's loop. Awaiting it
# from another loop (the tool worker thread) deadlocks on a cross-loop
# Future, so dispatch onto the gateway loop instead.
gateway_loop = getattr(runner, "_gateway_loop", None)
try:
current_loop = asyncio.get_running_loop()
@@ -562,28 +436,23 @@ async def _send_via_adapter(
return {"error": "Gateway loop is not running; cannot dispatch adapter send"}
from agent.async_utils import safe_schedule_threadsafe
fut = safe_schedule_threadsafe(
make_coro(), gateway_loop, logger=logger, log_message=log_message
)
make_coro(), gateway_loop, logger=logger, log_message=log_message)
if fut is None:
return {"error": "Gateway loop unavailable for send dispatch"}
# shield: a cancelled caller (agent interrupt) must not cancel the
# already-enqueued gateway send, or a retry would duplicate it.
# No timeout here — the adapter's request timeout and the outer
# _run_async timeout bound the wait.
# already-enqueued gateway send, or a retry would duplicate it. No timeout
# here — the adapter's request timeout and the outer _run_async bound the wait.
return await asyncio.shield(asyncio.wrap_future(fut))
if media_files:
return await _dispatch(
lambda: _send_live_adapter_media(
adapter, chat_id, chunk, media_files,
thread_id=thread_id, metadata=metadata, force_document=force_document,
),
"send_message: failed to schedule media send on gateway loop",
)
thread_id=thread_id, metadata=metadata, force_document=force_document),
"send_message: failed to schedule media send on gateway loop")
result = await _dispatch(
lambda: adapter.send(chat_id=chat_id, content=chunk, metadata=metadata),
"send_message: failed to schedule on gateway loop",
)
"send_message: failed to schedule on gateway loop")
except asyncio.CancelledError:
raise
except Exception as e:
@@ -599,43 +468,29 @@ async def _send_via_adapter(
entry = platform_registry.get(platform_name)
except Exception:
entry = None
if entry is not None and entry.standalone_sender_fn is not None:
try:
result = await entry.standalone_sender_fn(
pconfig,
chat_id,
chunk,
thread_id=thread_id,
media_files=media_files,
force_document=force_document,
)
except asyncio.CancelledError:
raise
except Exception as e:
logger.debug("Plugin standalone send for %s raised", platform_name, exc_info=True)
return {"error": f"Plugin standalone send failed: {_bounded_send_error(e)}"}
if isinstance(result, dict) and (result.get("success") or result.get("error")):
if result.get("error"):
return {**result, "error": _bounded_send_error(result["error"])}
return result
return {
"error": (
f"Plugin standalone send for '{platform_name}' returned an "
f"invalid result: expected a dict with 'success' or 'error' "
f"keys, got {type(result).__name__}"
)
}
return {
"error": (
if entry is None or entry.standalone_sender_fn is None:
return {"error": (
f"No live adapter for platform '{platform_name}'. Is the gateway "
f"running with this platform connected? For out-of-process delivery "
f"(e.g. cron in a separate process), the platform plugin must "
f"register a standalone_sender_fn on its PlatformEntry."
)
}
f"register a standalone_sender_fn on its PlatformEntry.")}
try:
result = await entry.standalone_sender_fn(
pconfig, chat_id, chunk,
thread_id=thread_id, media_files=media_files, force_document=force_document)
except asyncio.CancelledError:
raise
except Exception as e:
logger.debug("Plugin standalone send for %s raised", platform_name, exc_info=True)
return {"error": f"Plugin standalone send failed: {_bounded_send_error(e)}"}
if isinstance(result, dict) and (result.get("success") or result.get("error")):
if result.get("error"):
return {**result, "error": _bounded_send_error(result["error"])}
return result
return {"error": (
f"Plugin standalone send for '{platform_name}' returned an "
f"invalid result: expected a dict with 'success' or 'error' "
f"keys, got {type(result).__name__}")}
async def _send_chunks(chunks, send_one):
@@ -651,10 +506,10 @@ async def _send_chunks(chunks, send_one):
def _platform_max_length(platform):
"""Max message length for chunking: the adapter constant for Signal (its
raw JSON-RPC path never sees SignalAdapter's own chunking, so signal-cli
would reject long sends whole), the registry's ``max_message_length`` for
plugins (Slack, Feishu, ...), else None (no chunking)."""
"""Max message length for chunking: the adapter constant for Signal (its raw
JSON-RPC path never sees SignalAdapter's own chunking, so signal-cli would reject
long sends whole), the registry's ``max_message_length`` for plugins (Slack,
Feishu, ...), else None (no chunking)."""
from gateway.config import Platform
if platform == Platform.SIGNAL:
@@ -673,20 +528,18 @@ def _platform_max_length(platform):
return None
# Plugin platforms whose media (and, for Discord, all) sends go straight to the
# registry ``standalone_sender_fn`` — bypassing the live adapter on purpose:
# Discord's plugin ``_standalone_send`` handles forum channels, threads and
# multipart uploads (historically the only Discord path); Slack uploads via
# files_upload_v2; WhatsApp posts each file to the Baileys bridge /send-media so
# images/videos/audio arrive as native bubbles, not documents.
# Plugin platforms whose media (and, for Discord, all) sends go straight to the registry
# ``standalone_sender_fn`` — bypassing the live adapter on purpose: Discord's plugin
# ``_standalone_send`` handles forum channels, threads and multipart uploads
# (historically the only Discord path); Slack uploads via files_upload_v2; WhatsApp
# posts each file to the Baileys bridge /send-media so media arrive as native bubbles.
# platform -> (error label, run discover_plugins first, caption-capable,
# media_files sentinel for non-final chunks, forward force_document)
_PLUGIN_STANDALONE_MEDIA = {
"discord": ("Discord", False, True, [], False),
"feishu": ("Feishu", True, False, None, False),
"slack": ("Slack", True, True, [], False),
"whatsapp": ("WhatsApp", True, True, None, True),
}
"whatsapp": ("WhatsApp", True, True, None, True)}
async def _send_plugin_standalone(
@@ -702,36 +555,35 @@ async def _send_plugin_standalone(
if captionable:
# Cap on the platform's own message limit so the caption is deliverable.
caption, _ = _media_caption_split(
message, media_files, max_caption_len=(max_len or _DEFAULT_CAPTION_LIMIT)
)
message, media_files, max_caption_len=(max_len or _DEFAULT_CAPTION_LIMIT))
if caption is not None:
return await sender(
pconfig, chat_id, "", thread_id=thread_id, media_files=media_files,
caption=caption, **extra,
)
caption=caption, **extra)
return await _send_chunks(
chunks,
lambda chunk, is_last: sender(
pconfig, chat_id, chunk, thread_id=thread_id,
media_files=media_files if is_last else empty_media, **extra,
),
)
media_files=media_files if is_last else empty_media, **extra))
# Native-media chunked routes for built-in platforms; media rides on the final
# chunk, non-final chunks get the platform's empty-media sentinel.
# Native-media chunked routes for built-in platforms; media rides on the final chunk,
# non-final chunks get the platform's empty-media sentinel.
# platform -> (media required, empty-media sentinel,
# sender(platform, pconfig, chat_id, chunk, media, thread_id, force_document))
# Matrix: ALL sends go through the native adapter so text is encrypted in E2EE
# rooms too (the raw-HTTP standalone path is not encryption-aware).
# Matrix: ALL sends go through the native adapter so text is encrypted in E2EE rooms
# too (the raw-HTTP standalone path is not encryption-aware).
# Signal: attachments ride the JSON-RPC ``attachments`` param.
# Yuanbao: media needs the running gateway adapter's WebSocket.
# Slack (text; media was intercepted above): prefer the live adapter — it is
# multi-workspace aware and honors adapter gates like ignored_channels, while
# the standalone Web-API path may only have a token list — else the plugin's
# standalone sender, keeping MEDIA delivery on the cron fallback.
# Slack (text; media was intercepted above): prefer the live adapter — multi-workspace
# aware and honors adapter gates like ignored_channels, while the standalone Web-API
# path may only have a token list — else the plugin's standalone sender.
# WeCom: native media only through the live gateway adapter.
# Names resolve at call time so tests can monkeypatch e.g. ``_send_signal``.
def _via_adapter_route(p, pc, cid, chunk, media, tid, fd):
return _send_via_adapter(p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd)
_CHUNKED_ROUTES = {
"matrix": (False, [], lambda p, pc, cid, chunk, media, tid, fd: _send_matrix_via_adapter(
pc, cid, chunk, media_files=media, thread_id=tid)),
@@ -739,26 +591,22 @@ _CHUNKED_ROUTES = {
pc.extra, cid, chunk, media_files=media)),
"yuanbao": (True, None, lambda p, pc, cid, chunk, media, tid, fd: _send_yuanbao(
cid, chunk, media_files=media)),
"slack": (False, [], lambda p, pc, cid, chunk, media, tid, fd: _send_via_adapter(
p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd)),
"wecom": (True, None, lambda p, pc, cid, chunk, media, tid, fd: _send_via_adapter(
p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd)),
}
"slack": (False, [], _via_adapter_route),
"wecom": (True, None, _via_adapter_route)}
# Text-only senders for built-in platforms (generic path; media is dropped
# with a warning). Signature: (pconfig, chat_id, chunk, thread_id) -> result.
def _registry_text_sender(name):
return lambda pc, cid, chunk, tid: _registry_standalone_send(name, pc, cid, chunk, tid)
# Text-only senders for built-in platforms (generic path; media is dropped with a
# warning). Signature: (pconfig, chat_id, chunk, thread_id) -> result.
_TEXT_SENDERS = {
"whatsapp": lambda pc, cid, chunk, tid: _registry_standalone_send("whatsapp", pc, cid, chunk, tid),
**{name: _registry_text_sender(name) for name in ("whatsapp", "email", "sms", "dingtalk", "feishu", "wecom")},
"signal": lambda pc, cid, chunk, tid: _send_signal(pc.extra, cid, chunk),
"email": lambda pc, cid, chunk, tid: _registry_standalone_send("email", pc, cid, chunk, tid),
"sms": lambda pc, cid, chunk, tid: _registry_standalone_send("sms", pc, cid, chunk, tid),
"dingtalk": lambda pc, cid, chunk, tid: _registry_standalone_send("dingtalk", pc, cid, chunk, tid),
"feishu": lambda pc, cid, chunk, tid: _registry_standalone_send("feishu", pc, cid, chunk, tid),
"wecom": lambda pc, cid, chunk, tid: _registry_standalone_send("wecom", pc, cid, chunk, tid),
"bluebubbles": lambda pc, cid, chunk, tid: _send_bluebubbles(pc.extra, cid, chunk),
"qqbot": lambda pc, cid, chunk, tid: _send_qqbot(pc, cid, chunk),
"yuanbao": lambda pc, cid, chunk, tid: _send_yuanbao(cid, chunk),
}
"yuanbao": lambda pc, cid, chunk, tid: _send_yuanbao(cid, chunk)}
_MEDIA_PLATFORMS_NOTE = "telegram, discord, matrix, weixin, signal, yuanbao, feishu, whatsapp and slack"
@@ -766,12 +614,11 @@ _MEDIA_PLATFORMS_NOTE = "telegram, discord, matrix, weixin, signal, yuanbao, fei
async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None, force_document=False, args=None):
"""Route a message to the appropriate platform sender.
Long messages are chunked with the adapters' smart splitter (code-block
aware, part indicators). Branch order matters: Weixin first (its native
helper must not be blocked by unrelated optional imports such as
lark-oapi's heavy Feishu path), Telegram (chunks itself), plugin
standalone media routes, native chunked routes, then the generic text
path that drops media with a warning.
Long messages are chunked with the adapters' smart splitter (code-block aware,
part indicators). Branch order matters: Weixin first (its native helper must not
be blocked by unrelated optional imports such as lark-oapi's heavy Feishu path),
Telegram (chunks itself), plugin standalone media routes, native chunked routes,
then the generic text path that drops media with a warning.
"""
from gateway.config import Platform
@@ -781,20 +628,14 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
if platform == Platform.WEIXIN:
return await _send_weixin(pconfig, chat_id, message, media_files=media_files)
# Telegram chunks internally on the *formatted* text in UTF-16 units
# (MarkdownV2/HTML escaping inflates length), so it gets the whole
# message; media attaches after all text chunks.
# Telegram chunks internally on the *formatted* text in UTF-16 units (MarkdownV2/HTML
# escaping inflates length), so it gets the whole message; media attaches after all text.
if platform == Platform.TELEGRAM:
disable_link_previews = bool(getattr(pconfig, "extra", {}) and pconfig.extra.get("disable_link_previews"))
return await _send_telegram(
pconfig.token,
chat_id,
message,
media_files=media_files,
thread_id=thread_id,
disable_link_previews=disable_link_previews,
force_document=force_document,
)
pconfig.token, chat_id, message,
media_files=media_files, thread_id=thread_id,
disable_link_previews=disable_link_previews, force_document=force_document)
from gateway.platforms.base import BasePlatformAdapter
@@ -804,33 +645,27 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
if platform_name == "discord" or (media_files and platform_name in _PLUGIN_STANDALONE_MEDIA):
return await _send_plugin_standalone(
platform_name, pconfig, chat_id, message, chunks, media_files,
thread_id=thread_id, max_len=max_len, force_document=force_document,
)
thread_id=thread_id, max_len=max_len, force_document=force_document)
route = _CHUNKED_ROUTES.get(platform_name)
if route is not None and (media_files or not route[0]):
_, empty_media, sender = route
return await _send_chunks(chunks, lambda chunk, is_last: sender(
platform, pconfig, chat_id, chunk,
media_files if is_last else empty_media, thread_id, force_document,
))
media_files if is_last else empty_media, thread_id, force_document))
# --- Generic path: text only. Buzz is a plugin platform with verified
# native media delivery through _send_via_adapter (media-only sends
# included), so it is exempt from the media error/warning.
if media_files and platform_name != "buzz" and not message.strip():
return {
"error": (
f"send_message MEDIA delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}; "
f"target {platform_name} had only media attachments"
)
}
# --- Generic path: text only. Buzz is a plugin platform with verified native media
# delivery through _send_via_adapter (media-only sends included), so it is exempt
# from the media error/warning.
warning = None
if media_files and platform_name != "buzz":
if not message.strip():
return {"error": (
f"send_message MEDIA delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}; "
f"target {platform_name} had only media attachments")}
warning = (
f"MEDIA attachments were omitted for {platform_name}; "
f"native send_message media delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}"
)
f"native send_message media delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}")
text_sender = _TEXT_SENDERS.get(platform_name)
if text_sender is not None:
@@ -852,27 +687,11 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
except Exception as e:
return {"error": f"Plugin send_message handler failed: {e}"}
# Plugin platform: live gateway adapter if available, else standalone_sender_fn.
send_one = lambda chunk, is_last: _send_via_adapter( # noqa: E731
platform, pconfig, chat_id, chunk, thread_id=thread_id,
media_files=media_files if is_last else [], force_document=force_document,
)
send_one = lambda chunk, is_last: _via_adapter_route( # noqa: E731
platform, pconfig, chat_id, chunk, media_files if is_last else [], thread_id, force_document)
last_result = await _send_chunks(chunks, send_one)
if (
warning
and isinstance(last_result, dict)
and last_result.get("success")
and not last_result.get("media_delivered")
):
if (warning and isinstance(last_result, dict) and last_result.get("success")
and not last_result.get("media_delivered")):
last_result["warnings"] = [*last_result.get("warnings", []), warning]
return last_result
# --- Registry ---
from tools.registry import tool_error
# NOTE: ``send_message`` is intentionally NOT registered as an agent-callable
# model tool (the agent must not fire cross-platform messages on its own).
# The send engine here is the shared transport for cron delivery, the
# ``hermes send`` CLI, the gateway kanban notifier and the opt-in MCP server,
# which import the helpers directly.