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

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

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:41:03 -07:00
parent adada20660 47117f51ba
commit a833865ecf
3 changed files with 411 additions and 794 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/platforms/line/adapter.py
+60 -85
View File
@@ -379,6 +379,7 @@ _OUTBOUND_MEDIA = {
# Inbound media kinds → cached file extension.
_INBOUND_MEDIA_EXT = {"image": ".jpg", "audio": ".m4a", "video": ".mp4", "file": ".bin"}
_LIFECYCLE_EVENTS = frozenset({"follow", "unfollow", "join", "leave"})
class LineAdapter(BasePlatformAdapter):
@@ -557,7 +558,7 @@ class LineAdapter(BasePlatformAdapter):
await self._handle_message_event(event)
elif event_type == "postback":
await self._handle_postback_event(event)
elif event_type in {"follow", "unfollow", "join", "leave"}:
elif event_type in _LIFECYCLE_EVENTS:
logger.info("LINE: lifecycle event %s from %s", event_type, source)
else:
logger.debug("LINE: ignoring event type %r", event_type)
@@ -580,8 +581,7 @@ class LineAdapter(BasePlatformAdapter):
local_path, media_type = await self._download_media(
message_id, msg_type, filename=msg.get("fileName") or msg.get("file_name"))
if local_path:
media_urls.append(local_path)
media_types.append(media_type)
media_urls, media_types = [local_path], [media_type]
text = f"[{msg_type}]"
elif msg_type == "sticker":
keywords = msg.get("keywords") or []
@@ -593,61 +593,48 @@ class LineAdapter(BasePlatformAdapter):
if chat_type == "dm" and self._client: # best-effort typing indicator (DM only)
asyncio.create_task(self._client.loading(chat_id))
source_obj = self.build_source(
chat_id=chat_id, chat_type=chat_type, user_id=user_id, user_name=user_id, chat_name=chat_id
)
chat_id=chat_id, chat_type=chat_type, user_id=user_id, user_name=user_id, chat_name=chat_id)
await self.handle_message(MessageEvent(
text=text, message_type=_LINE_MESSAGE_TYPES.get(msg_type, MessageType.TEXT), source=source_obj,
raw_message=event, message_id=message_id, media_urls=media_urls, media_types=media_types,
))
raw_message=event, message_id=message_id, media_urls=media_urls, media_types=media_types))
async def _handle_postback_event(self, event: Dict[str, Any]) -> None:
"""User tapped the slow-LLM postback button — deliver cached payload."""
postback = event.get("postback") or {}
data = postback.get("data", "") or ""
"""User tapped the slow-LLM postback button — deliver the cached payload. READY replies (push
fallback) and ERROR replies settle the entry; DELIVERED / PENDING just re-issue their notice."""
reply_token = event.get("replyToken", "")
chat_id, _ = _resolve_chat(event.get("source") or {})
try:
parsed = json.loads(data)
parsed = json.loads((event.get("postback") or {}).get("data", "") or "")
except (TypeError, json.JSONDecodeError):
return
if parsed.get("action") != "show_response":
return
request_id = parsed.get("request_id", "")
if not request_id:
return
entry = self._cache.get(request_id)
request_id = parsed.get("request_id", "") if parsed.get("action") == "show_response" else ""
entry = self._cache.get(request_id) if request_id else None
if not self._client or not reply_token or not entry:
return
def _settle() -> None:
self._cache.mark_delivered(request_id)
self._pending_buttons.pop(chat_id, None)
if entry.state is State.READY:
state = entry.state
if state is State.READY:
messages = _text_messages(str(entry.payload or ""))
try:
await self._client.reply(reply_token, messages)
_settle()
except Exception as exc:
elif state is State.ERROR:
messages = [_text_message(str(entry.payload or self.interrupted_text))]
else:
messages = [_text_message(self.delivered_text if state is State.DELIVERED else self.pending_text)]
try:
await self._client.reply(reply_token, messages)
except Exception as exc:
if state is State.READY:
logger.warning("LINE: postback reply failed (%s); falling back to push", exc)
try:
await self._client.push(chat_id, messages)
_settle()
except Exception as exc2:
logger.error("LINE: postback push fallback failed: %s", exc2)
elif entry.state is State.ERROR:
text = str(entry.payload or self.interrupted_text)
try:
await self._client.reply(reply_token, [_text_message(text)])
_settle()
except Exception as exc:
logger.warning("LINE: postback ERROR reply failed: %s", exc)
elif entry.state in (State.DELIVERED, State.PENDING): # "already replied" / re-issue wait notice
text = self.delivered_text if entry.state is State.DELIVERED else self.pending_text
try:
await self._client.reply(reply_token, [_text_message(text)])
except Exception:
pass
return
else:
if state is State.ERROR:
logger.warning("LINE: postback ERROR reply failed: %s", exc)
return
if state in (State.READY, State.ERROR):
self._cache.mark_delivered(request_id)
self._pending_buttons.pop(chat_id, None)
async def _download_media(
self, message_id: str, msg_type: str, *, filename: Optional[str] = None
@@ -692,9 +679,7 @@ class LineAdapter(BasePlatformAdapter):
def _consume_reply_token(self, chat_id: str) -> Tuple[str, bool]:
"""Consume a stashed reply token if present and unexpired → ``(token, used_reply)``."""
token, expires_at = self._reply_tokens.pop(chat_id, None) or ("", 0.0)
if not token or time.time() >= expires_at:
return "", False
return token, True
return (token, True) if token and time.time() < expires_at else ("", False)
async def send_typing(self, chat_id: str, metadata=None) -> None:
"""Trigger LINE's loading-animation indicator (DM only)."""
@@ -808,15 +793,14 @@ class LineAdapter(BasePlatformAdapter):
path is rechecked against allowed roots (tempdir, ``/tmp``→``/private/tmp`` on macOS, HERMES_HOME)."""
from aiohttp import web
token = request.match_info["token"]
entry = self._media_tokens.get(token)
if not entry:
file_path, expires_at = self._media_tokens.get(token) or ("", 0.0)
if not file_path:
return web.Response(status=404, text="not found")
file_path, expires_at = entry
if time.time() > expires_at:
self._media_tokens.pop(token, None)
return web.Response(status=410, text="gone")
path = Path(file_path)
if not path.exists() or not path.is_file():
if not path.is_file():
return web.Response(status=404, text="not found")
try:
from hermes_constants import get_hermes_home
@@ -828,8 +812,8 @@ class LineAdapter(BasePlatformAdapter):
if not any(resolved.is_relative_to(r) for r in allowed_roots):
logger.warning("LINE: refusing to serve outside allowed roots: %s", resolved)
return web.Response(status=403, text="forbidden")
content_type, _ = mimetypes.guess_type(str(path))
return web.FileResponse(path, headers={"Content-Type": content_type or "application/octet-stream"})
content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
return web.FileResponse(path, headers={"Content-Type": content_type})
async def send_image_file(
self, chat_id: str, image_path: str, caption: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None
@@ -841,9 +825,7 @@ class LineAdapter(BasePlatformAdapter):
if not url.lower().startswith("https://"):
return SendResult(success=False, error=f"LINE image URL must be HTTPS: {url}")
msgs: List[Dict[str, Any]] = [{"type": "image", "originalContentUrl": url, "previewImageUrl": url}]
if caption:
msgs.append(_text_message(caption))
return await self._send_messages(chat_id, msgs)
return await self._send_messages(chat_id, msgs + ([_text_message(caption)] if caption else []))
async def send_voice(
self, chat_id: str, audio_path: str, duration_ms: int = 1000, metadata: Optional[Dict[str, Any]] = None
@@ -851,8 +833,8 @@ class LineAdapter(BasePlatformAdapter):
path, err = self._check_media_file("audio", audio_path)
if err:
return err
url = self._serve_file(path)
return await self._send_messages(chat_id, [{"type": "audio", "originalContentUrl": url, "duration": int(duration_ms)}])
msg = {"type": "audio", "originalContentUrl": self._serve_file(path), "duration": int(duration_ms)}
return await self._send_messages(chat_id, [msg])
async def send_video(
self, chat_id: str, video_path: str, preview_path: Optional[str] = None,
@@ -873,10 +855,8 @@ class LineAdapter(BasePlatformAdapter):
except Exception:
_unlink_quietly(tmp.name)
raise
video_url = self._serve_file(path)
return await self._send_messages(
chat_id, [{"type": "video", "originalContentUrl": video_url, "previewImageUrl": preview_url}]
)
msg = {"type": "video", "originalContentUrl": self._serve_file(path), "previewImageUrl": preview_url}
return await self._send_messages(chat_id, [msg])
async def _send_messages(
self, chat_id: str, messages: List[Dict[str, Any]], *, force_push: bool = False, text: bool = False
@@ -887,32 +867,27 @@ class LineAdapter(BasePlatformAdapter):
return SendResult(success=False, error="LINE adapter not connected")
if not messages:
return SendResult(success=True, message_id=None)
first_batch = messages[:LINE_MAX_MESSAGES_PER_CALL]
rest = messages[LINE_MAX_MESSAGES_PER_CALL:]
n = LINE_MAX_MESSAGES_PER_CALL
batches = [messages[i:i + n] for i in range(0, len(messages), n)]
token, used_reply = self._consume_reply_token(chat_id)
start = 0
if used_reply and not force_push:
try:
await self._client.reply(token, first_batch)
first_batch = None
await self._client.reply(token, batches[0])
if text:
return SendResult(success=True, message_id=token)
start = 1
except Exception as exc:
logger.info("LINE: reply token rejected (%s); falling back to push", exc)
if first_batch is not None:
for i in range(start, len(batches)): # push the rest (reply token is single-use)
try:
await self._client.push(chat_id, first_batch)
await self._client.push(chat_id, batches[i])
except Exception as exc:
if text:
if i > 0:
logger.warning("LINE: push for follow-up batch failed: %s", exc)
elif text:
logger.error("LINE: push send failed: %s", exc)
return SendResult(success=False, error=str(exc))
while rest: # subsequent batches: always push (reply token is single-use)
batch = rest[:LINE_MAX_MESSAGES_PER_CALL]
rest = rest[LINE_MAX_MESSAGES_PER_CALL:]
try:
await self._client.push(chat_id, batch)
except Exception as exc:
logger.warning("LINE: push for follow-up batch failed: %s", exc)
return SendResult(success=False, error=str(exc))
return SendResult(success=True, message_id=None)
@@ -923,9 +898,13 @@ def _unlink_quietly(path: str) -> None:
pass
def _env_credentials_present() -> bool:
return bool(_get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") and _get_scoped_secret("LINE_CHANNEL_SECRET"))
def check_requirements() -> bool:
"""Plugin gate: require credentials AND aiohttp at runtime."""
if not (_get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") and _get_scoped_secret("LINE_CHANNEL_SECRET")):
if not _env_credentials_present():
return False
try:
import aiohttp # noqa: F401
@@ -935,8 +914,7 @@ def check_requirements() -> bool:
def validate_config(config) -> bool:
token, secret = _credentials(config)
return bool(token) and bool(secret)
return all(_credentials(config))
def is_connected(config) -> bool:
@@ -946,7 +924,7 @@ def is_connected(config) -> bool:
def _env_enablement() -> Optional[Dict[str, Any]]:
"""Seed PlatformConfig.extra from env-only setups so ``hermes status`` sees them."""
if not (_get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") and _get_scoped_secret("LINE_CHANNEL_SECRET")):
if not _env_credentials_present():
return None
seeded: Dict[str, Any] = {}
if os.getenv("LINE_PORT"):
@@ -957,17 +935,15 @@ def _env_enablement() -> Optional[Dict[str, Any]]:
for env, key in (("LINE_HOST", "host"), ("LINE_PUBLIC_URL", "public_url"), ("LINE_HOME_CHANNEL", "home_channel")):
if os.getenv(env):
seeded[key] = os.environ[env]
return seeded or {}
return seeded
async def _standalone_send(
pconfig, chat_id: str, message: str, *,
thread_id: Optional[str] = None, media_files: Optional[List[str]] = None, force_document: bool = False,
) -> Dict[str, Any]:
"""Out-of-process Push delivery for cron jobs detached from the gateway.
Always Push (no inbound event → no reply token). ``thread_id`` is ignored (LINE
has no threads); ``media_files`` can't be served without the webhook server.
"""
"""Out-of-process Push delivery for cron jobs detached from the gateway (no inbound event → no
reply token). ``thread_id`` is ignored (no threads); ``media_files`` need the webhook server."""
extra = getattr(pconfig, "extra", {}) or {}
token = _get_scoped_secret("LINE_CHANNEL_ACCESS_TOKEN") or extra.get("channel_access_token", "")
if not token or not chat_id:
@@ -1014,8 +990,7 @@ def interactive_setup() -> None:
_prompt("LINE_CHANNEL_SECRET", "Channel secret", secret=True)
_prompt("LINE_PUBLIC_URL", "Public HTTPS base URL (optional, e.g. https://my-tunnel.example.com)")
_prompt("LINE_ALLOWED_USERS", "Allowed user IDs (comma-separated; blank=skip)")
print("Done. Set the webhook URL in the LINE console to "
"<your-public-url>/line/webhook and enable 'Use webhook'.")
print("Done. Set the webhook URL in the LINE console to <your-public-url>/line/webhook and enable 'Use webhook'.")
def register(ctx) -> None:
plugins/platforms/simplex/adapter.py
+176 -405
View File
@@ -1,41 +1,12 @@
"""SimpleX Chat platform adapter (Hermes plugin).
"""SimpleX Chat adapter: persistent WebSocket to a simplex-chat daemon (``simplex-chat -p 5225``,
or ``docker run -p 5225:5225 simplexchat/simplex-chat-cli -p 5225``); JSON commands out, events in.
Connects to a simplex-chat daemon running in WebSocket mode. Inbound messages
arrive via a persistent WebSocket connection; outbound messages use the same
WebSocket with JSON commands. The plugin loader calls ``register(ctx)`` at
startup and the platform becomes available to ``gateway/run.py`` and
``tools/send_message_tool`` through the registry.
SimpleX chat daemon setup:
simplex-chat -p 5225 # start daemon on port 5225
# or via Docker:
# docker run -p 5225:5225 simplexchat/simplex-chat-cli -p 5225
Required environment variables:
SIMPLEX_WS_URL WebSocket URL of the daemon
(default: ws://127.0.0.1:5225)
Optional environment variables:
SIMPLEX_ALLOWED_USERS Comma-separated allowlist. Each entry may be
either a numeric contactId (stable across
renames; visible via `/contacts` in the CLI)
or a contact display name (what the SimpleX
UI shows). Both forms are accepted.
SIMPLEX_ALLOW_ALL_USERS Set 'true' to allow all contacts
SIMPLEX_AUTO_ACCEPT Set 'false' to disable contact-request auto-accept
(default: 'true')
SIMPLEX_GROUP_ALLOWED Comma-separated group IDs to monitor, or '*'
for any group. Omit to disable groups entirely.
SIMPLEX_HOME_CHANNEL Default contact/group ID for cron delivery
SIMPLEX_HOME_CHANNEL_NAME Human label for the home channel
HERMES_SIMPLEX_TEXT_BATCH_DELAY
Quiet-period seconds (default: 0.8) used to
concatenate rapid-fire inbound text messages
into a single MessageEvent.
``websockets`` is imported lazily so the plugin stays discoverable (and
``hermes setup`` can describe it) when the package is missing;
``check_requirements()`` returns False until it is present.
Env: SIMPLEX_WS_URL (required; default ws://127.0.0.1:5225) · SIMPLEX_ALLOWED_USERS (numeric
contactIds — stable across renames, see ``/contacts`` — or display names) · SIMPLEX_ALLOW_ALL_USERS ·
SIMPLEX_AUTO_ACCEPT ('false' disables contact-request auto-accept; default true) ·
SIMPLEX_GROUP_ALLOWED (group IDs or '*'; omit to ignore groups) · SIMPLEX_HOME_CHANNEL[_NAME] ·
HERMES_SIMPLEX_TEXT_BATCH_DELAY (quiet seconds, default 0.8, merging rapid-fire inbound text).
``websockets`` is imported lazily so the plugin stays discoverable when the package is missing.
"""
import asyncio
@@ -50,9 +21,11 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import unquote
from gateway.platforms._shared import get_scoped_secret as _get_scoped_secret
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult
from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult, cache_image_from_url
logger = logging.getLogger(__name__)
@@ -67,6 +40,7 @@ _CORR_PREFIX = "hermes-"
_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
_AUDIO_EXTS = {".mp3", ".wav", ".ogg", ".m4a", ".aac", ".opus"}
_VOICE_TAG_EXTS = {".ogg", ".mp3", ".wav", ".m4a", ".opus"} # MEDIA: tags sent as voice notes
def _parse_comma_list(value: str) -> List[str]:
@@ -90,18 +64,16 @@ def _is_audio_ext(ext: str) -> bool:
return ext.lower() in _AUDIO_EXTS
def _send_cmd(chat_id: str, items: list) -> str:
"""Build a structured ``/_send`` command addressing *chat_id* by ID.
def _display_name(obj: dict, profile_key: str, default: str = "") -> str:
"""``localDisplayName`` falling back to the nested profile's ``displayName``."""
return obj.get("localDisplayName", "") or (obj.get(profile_key, {}) or {}).get("displayName", default)
The structured json form is used (rather than ``@name text`` / ``#[id]``)
because the daemon parses the bare syntax as a display-name lookup and
silently drops messages when the name doesn't resolve; json.dumps also
escapes newlines/special chars correctly.
"""
composed = json.dumps(items)
if chat_id.startswith("group:"):
return f"/_send #{chat_id[6:]} json {composed}"
return f"/_send @{chat_id} json {composed}"
def _send_cmd(chat_id: str, items: list) -> str:
"""Structured ``/_send <target> json …`` addressing *chat_id* by ID. The bare ``@name`` / ``#[id]``
syntax is a display-name lookup that silently drops unresolved names; json.dumps escapes text."""
target = f"#{chat_id[6:]}" if chat_id.startswith("group:") else f"@{chat_id}"
return f"/_send {target} json {json.dumps(items)}"
async def _cancel_task(task: Optional[asyncio.Task]) -> None:
@@ -123,18 +95,14 @@ class SimplexAdapter(BasePlatformAdapter):
extra = getattr(config, "extra", {}) or {}
self.ws_url = extra.get("ws_url", "ws://127.0.0.1:5225").rstrip("/")
# Auto-accept is on by default; env wins over the ``_env_enablement`` seed.
env_auto = _get_scoped_secret("SIMPLEX_AUTO_ACCEPT")
if env_auto is not None:
self.auto_accept = env_auto.strip().lower() not in {"0", "false", "no", ""}
else:
self.auto_accept = bool(extra.get("auto_accept", True))
# Without SIMPLEX_GROUP_ALLOWED, group messages are ignored entirely
# (safer default). ``*`` accepts any group.
group_allowed_str = (_get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "")
or extra.get("group_allowed", ""))
# Without SIMPLEX_GROUP_ALLOWED group messages are ignored (safer default); ``*`` = any group.
group_allowed_str = _get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "") or extra.get("group_allowed", "")
self.group_allow_from = set(_parse_comma_list(group_allowed_str))
self._ws = None # websockets connection
@@ -142,31 +110,18 @@ class SimplexAdapter(BasePlatformAdapter):
self._health_task: Optional[asyncio.Task] = None
self._running = False
self._last_ws_activity = 0.0
# Cosmetic echo filter: corrIds we minted, bounded.
self._pending_corr_ids: set = set()
self._pending_corr_ids: set = set() # cosmetic echo filter: corrIds we minted, bounded
self._max_pending_corr = 200
# File transfers awaiting rcvFileComplete (keyed by fileId).
self._pending_file_transfers: Dict[int, dict] = {}
# Futures for commands whose responses we actually await (``_send_command``).
self._pending_responses: Dict[str, asyncio.Future] = {}
self._pending_file_transfers: Dict[int, dict] = {} # awaiting rcvFileComplete, by fileId
self._pending_responses: Dict[str, asyncio.Future] = {} # awaited command replies
self._corr_counter = 0
# Text batching state consumed by BasePlatformAdapter._enqueue_text_event.
self._text_batch_delay = float(os.getenv("HERMES_SIMPLEX_TEXT_BATCH_DELAY", "0.8"))
self._pending_text_batches: Dict[str, MessageEvent] = {}
self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
logger.info(
"SimpleX adapter initialized: url=%s auto_accept=%s groups=%s",
self.ws_url, self.auto_accept, "enabled" if self.group_allow_from else "disabled",
)
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
self.ws_url, self.auto_accept, "enabled" if self.group_allow_from else "disabled")
async def connect(self, *, is_reconnect: bool = False) -> bool:
"""Connect to the simplex-chat daemon and start the WebSocket listener."""
@@ -179,22 +134,17 @@ class SimplexAdapter(BasePlatformAdapter):
if not self.ws_url:
logger.error("SimpleX: SIMPLEX_WS_URL is required")
return False
# Quick connectivity check — open and immediately close.
try:
try: # quick connectivity check — open and immediately close
async with _wsclient.connect(self.ws_url, open_timeout=10):
pass
except Exception as e:
logger.error("SimpleX: cannot reach daemon at %s: %s", self.ws_url, e)
return False
self._running = True
self._last_ws_activity = time.time()
self._ws_task = asyncio.create_task(self._ws_listener())
self._health_task = asyncio.create_task(self._health_monitor())
if hasattr(self, "_mark_connected"):
self._mark_connected()
self._mark_connected()
logger.info("SimpleX: connected to %s", self.ws_url)
self._wire_plugin_handlers(None)
return True
@@ -204,33 +154,22 @@ class SimplexAdapter(BasePlatformAdapter):
self._running = False
await _cancel_task(self._ws_task)
await _cancel_task(self._health_task)
if self._ws:
try:
await self._ws.close()
except Exception:
pass
self._ws = None
for task in list(self._pending_text_batch_tasks.values()):
if not task.done():
task.cancel()
for pending in (self._pending_text_batch_tasks.values(), self._pending_responses.values()):
for item in list(pending):
if not item.done():
item.cancel()
self._pending_text_batch_tasks.clear()
self._pending_text_batches.clear()
for fut in self._pending_responses.values():
if not fut.done():
fut.cancel()
self._pending_responses.clear()
if hasattr(self, "_mark_disconnected"):
self._mark_disconnected()
self._mark_disconnected()
logger.info("SimpleX: disconnected")
# ------------------------------------------------------------------
# WebSocket listener / health
# ------------------------------------------------------------------
async def _ws_listener(self) -> None:
"""Maintain a persistent WebSocket connection to the daemon."""
import websockets as _wsclient
@@ -240,14 +179,11 @@ class SimplexAdapter(BasePlatformAdapter):
while self._running:
try:
logger.debug("SimpleX WS: connecting to %s", self.ws_url)
async with _wsclient.connect(
self.ws_url, ping_interval=20, ping_timeout=20, close_timeout=10
) as ws:
async with _wsclient.connect(self.ws_url, ping_interval=20, ping_timeout=20, close_timeout=10) as ws:
self._ws = ws
backoff = WS_RETRY_DELAY_INITIAL
self._last_ws_activity = time.time()
logger.info("SimpleX WS: connected")
async for raw in ws:
if not self._running:
break
@@ -262,26 +198,19 @@ class SimplexAdapter(BasePlatformAdapter):
break
except ConnectionClosed as e:
if self._running:
logger.warning("SimpleX WS: connection closed: %s (reconnecting in %.0fs)",
e, backoff)
logger.warning("SimpleX WS: connection closed: %s (reconnecting in %.0fs)", e, backoff)
except Exception as e:
if self._running:
logger.warning("SimpleX WS: unexpected error: %s (reconnecting in %.0fs)",
e, backoff)
logger.warning("SimpleX WS: unexpected error: %s (reconnecting in %.0fs)", e, backoff)
finally:
self._ws = None
if self._running:
await asyncio.sleep(backoff + backoff * 0.2 * random.random())
backoff = min(backoff * 2, WS_RETRY_DELAY_MAX)
async def _health_monitor(self) -> None:
"""Observe WebSocket idleness without reconnecting healthy quiet links.
simplex-chat can legitimately stay application-silent for long periods;
the websockets client already sends protocol pings, so idleness is only
logged — reconnecting on it causes needless churn.
"""
"""Log (never reconnect on) WebSocket idleness: simplex-chat legitimately stays
application-silent for long periods and the client already sends protocol pings."""
while self._running:
await asyncio.sleep(HEALTH_CHECK_INTERVAL)
if not self._running:
@@ -290,83 +219,74 @@ class SimplexAdapter(BasePlatformAdapter):
if elapsed > HEALTH_CHECK_STALE_THRESHOLD:
logger.debug("SimpleX: WS application-idle for %.0fs", elapsed)
# ------------------------------------------------------------------
# Inbound event handling
# ------------------------------------------------------------------
async def _handle_event(self, event: dict) -> None:
"""Dispatch a daemon event to the appropriate handler."""
# Messages are usually {"corrId": ..., "resp": {"type": ...}}, but some
# daemons put the response fields at top level — normalize both.
# Usually {"corrId": ..., "resp": {"type": ...}}, but some daemons put the
# response fields at top level — normalize both.
resp = event.get("resp") if isinstance(event.get("resp"), dict) else event
corr_id = event.get("corrId")
if corr_id and corr_id in self._pending_responses:
fut = self._pending_responses.pop(corr_id)
if not fut.done():
fut.set_result(resp)
return
# Cosmetic echo filter: prefixed corrIds are ours but weren't awaited.
if corr_id and isinstance(corr_id, str) and corr_id.startswith(_CORR_PREFIX):
self._pending_corr_ids.discard(corr_id)
self._pending_corr_ids.discard(corr_id) # our own un-awaited echo
return
resp_type = resp.get("type") or event.get("type", "")
if resp_type == "contactRequest" and self.auto_accept:
contact_req = resp.get("contactRequest", {}) or {}
contact_req_id = contact_req.get("contactRequestId")
if contact_req_id is not None:
logger.info("SimpleX: auto-accepting contact request %s",
_redact_id(str(contact_req_id)))
await self._send_command(f"/accept {contact_req_id}")
return
# simplex fires rcvFileDescrReady before newChatItems for some (XFTP)
# files; start the download now, the chat item arrives later.
if resp_type == "rcvFileDescrReady":
rcv_file = resp.get("rcvFileTransfer", {}) or {}
file_id = rcv_file.get("fileId") if isinstance(rcv_file, dict) else None
if file_id is not None:
logger.debug("SimpleX: rcvFileDescrReady for fileId=%s — sending /freceive",
file_id)
await self._send_fire_and_forget(f"/freceive {file_id}")
return
if resp_type == "newChatItems":
chat_items = resp.get("chatItems", []) or []
if not isinstance(chat_items, list):
chat_items = [chat_items]
for item in chat_items:
await self._safe_handle_chat_item(item, "SimpleX: error processing chat item")
return
# Singular variant — some daemon versions emit this
if resp_type == "newChatItem":
await self._safe_handle_chat_item(resp, "SimpleX: error processing chat item")
return
# File transfer completion — deliver any deferred chat item
if resp_type == "rcvFileComplete":
chat_item_data = (resp.get("chatItem", {}) or {}).get("chatItem", {}) or {}
file_info = chat_item_data.get("file", {}) or {}
file_id = file_info.get("fileId") if isinstance(file_info, dict) else None
if file_id is not None and file_id in self._pending_file_transfers:
pending = self._pending_file_transfers.pop(file_id)
file_source = file_info.get("fileSource", {}) or {}
file_path = file_source.get("filePath") if isinstance(file_source, dict) else None
if file_path:
pending_item_data = pending.get("chatItem", {}) or {}
pending_item_data.setdefault("file", {})["fileSource"] = {"filePath": file_path}
pending["chatItem"] = pending_item_data
await self._safe_handle_chat_item(
pending, "SimpleX: error processing deferred file message")
return
if resp_type:
handler = self._EVENT_HANDLERS.get(resp_type)
if handler is not None:
await handler(self, resp)
elif resp_type:
logger.debug("SimpleX: unhandled event type: %s", resp_type)
async def _on_contact_request(self, resp: dict) -> None:
if not self.auto_accept:
logger.debug("SimpleX: unhandled event type: %s", "contactRequest")
return
contact_req_id = (resp.get("contactRequest", {}) or {}).get("contactRequestId")
if contact_req_id is not None:
logger.info("SimpleX: auto-accepting contact request %s", _redact_id(str(contact_req_id)))
await self._send_command(f"/accept {contact_req_id}")
async def _on_rcv_file_descr_ready(self, resp: dict) -> None:
"""XFTP files fire this before newChatItems; start the download now, the chat item arrives later."""
rcv_file = resp.get("rcvFileTransfer", {}) or {}
file_id = rcv_file.get("fileId") if isinstance(rcv_file, dict) else None
if file_id is not None:
logger.debug("SimpleX: rcvFileDescrReady for fileId=%s — sending /freceive", file_id)
await self._send_fire_and_forget(f"/freceive {file_id}")
async def _on_new_chat_items(self, resp: dict) -> None:
chat_items = resp.get("chatItems", []) or []
for item in chat_items if isinstance(chat_items, list) else [chat_items]:
await self._safe_handle_chat_item(item, "SimpleX: error processing chat item")
async def _on_new_chat_item(self, resp: dict) -> None:
"""Singular variant emitted by some daemon versions."""
await self._safe_handle_chat_item(resp, "SimpleX: error processing chat item")
async def _on_rcv_file_complete(self, resp: dict) -> None:
"""Deliver a chat item deferred until its file transfer completed."""
chat_item_data = (resp.get("chatItem", {}) or {}).get("chatItem", {}) or {}
file_info = chat_item_data.get("file", {}) or {}
file_id = file_info.get("fileId") if isinstance(file_info, dict) else None
if file_id is None or file_id not in self._pending_file_transfers:
return
pending = self._pending_file_transfers.pop(file_id)
file_source = file_info.get("fileSource", {}) or {}
file_path = file_source.get("filePath") if isinstance(file_source, dict) else None
if file_path:
pending_item_data = pending.get("chatItem", {}) or {}
pending_item_data.setdefault("file", {})["fileSource"] = {"filePath": file_path}
pending["chatItem"] = pending_item_data
await self._safe_handle_chat_item(pending, "SimpleX: error processing deferred file message")
_EVENT_HANDLERS = {
"contactRequest": _on_contact_request, "rcvFileDescrReady": _on_rcv_file_descr_ready,
"newChatItems": _on_new_chat_items, "newChatItem": _on_new_chat_item, "rcvFileComplete": _on_rcv_file_complete,
}
async def _safe_handle_chat_item(self, item: dict, err_msg: str) -> None:
try:
await self._handle_chat_item(item)
@@ -381,17 +301,13 @@ class SimplexAdapter(BasePlatformAdapter):
meta = chat_item_data.get("meta", {}) or {}
content = chat_item_data.get("content", {}) or {}
msg_content = content.get("msgContent", {}) or {}
# Filter out our own messages
item_direction = chat_item_data.get("chatDir", {}) or {}
direction_type = item_direction.get("type", "") if isinstance(item_direction, dict) else ""
if direction_type in ("directSnd", "groupSnd"):
if direction_type in ("directSnd", "groupSnd"): # our own messages
return
content_type = content.get("type", "") if isinstance(content, dict) else ""
if content_type != "rcvMsgContent":
return
text = ""
msg_type_str = msg_content.get("type", "") if isinstance(msg_content, dict) else ""
if msg_type_str in ("text", "file", "image", "voice", "link", "video"):
@@ -399,36 +315,28 @@ class SimplexAdapter(BasePlatformAdapter):
if not text and msg_type_str not in ("image", "file", "voice"):
return
sender_id = sender_name = chat_id = ""
is_group = False
is_group = chat_type == "group"
if chat_type == "direct":
contact = chat_info.get("contact", {}) or {}
sender_id = str(contact.get("contactId", ""))
sender_name = (contact.get("localDisplayName", "")
or contact.get("profile", {}).get("displayName", ""))
chat_id = sender_id
elif chat_type == "group":
sender_id = chat_id = str(contact.get("contactId", ""))
sender_name = chat_name = _display_name(contact, "profile")
elif is_group:
group_info = chat_info.get("groupInfo", {}) or {}
group_id = str(group_info.get("groupId", ""))
chat_id = f"group:{group_id}"
is_group = True
member = item_direction.get("groupMember", {}) or {}
sender_id = str(member.get("memberId", ""))
sender_name = (member.get("localDisplayName", "")
or member.get("memberProfile", {}).get("displayName", ""))
if self.group_allow_from:
if "*" not in self.group_allow_from and group_id not in self.group_allow_from:
logger.debug("SimpleX: group %s not in allowlist", _redact_id(group_id))
return
else:
sender_name = _display_name(member, "memberProfile")
chat_name = _display_name(group_info, "groupProfile", chat_id)
if not self.group_allow_from:
logger.debug("SimpleX: ignoring group message (no SIMPLEX_GROUP_ALLOWED)")
return
if "*" not in self.group_allow_from and group_id not in self.group_allow_from:
logger.debug("SimpleX: group %s not in allowlist", _redact_id(group_id))
return
else:
logger.debug("SimpleX: unhandled chat type: %s", chat_type)
return
if not sender_id:
logger.debug("SimpleX: ignoring message with no sender")
return
@@ -440,22 +348,17 @@ class SimplexAdapter(BasePlatformAdapter):
if file_info and isinstance(file_info, dict):
file_source = file_info.get("fileSource", {}) or {}
file_path = file_source.get("filePath") if isinstance(file_source, dict) else None
file_name = file_info.get("fileName", "")
file_id = file_info.get("fileId")
ext = Path(file_path).suffix.lower() if file_path else ""
if not ext and file_name:
ext = Path(file_name).suffix.lower()
# Voice notes typically arrive before the file finishes downloading;
# defer until rcvFileComplete. /freceive gets no corrId reply, so
# awaiting one would block the event loop.
if not ext and file_info.get("fileName", ""):
ext = Path(file_info["fileName"]).suffix.lower()
# Voice notes typically arrive before the file finishes downloading; defer until
# rcvFileComplete. /freceive gets no corrId reply, so awaiting one would block the loop.
if not file_path and _is_audio_ext(ext) and file_id is not None:
logger.info("SimpleX: voice file %d not yet received, accepting transfer", file_id)
self._pending_file_transfers[file_id] = chat_item
await self._send_fire_and_forget(f"/freceive {file_id}")
return
if file_path:
if _is_image_ext(ext):
mime = f"image/{ext.lstrip('.')}"
@@ -466,53 +369,34 @@ class SimplexAdapter(BasePlatformAdapter):
media_urls.append(file_path)
media_types.append(mime)
chat_name = sender_name
if is_group:
chat_name = group_info.get("localDisplayName", "") or group_info.get(
"groupProfile", {}
).get("displayName", chat_id)
source = self.build_source(
chat_id=chat_id, chat_name=chat_name, chat_type="group" if is_group else "dm",
user_id=sender_id, user_name=sender_name or sender_id,
)
user_id=sender_id, user_name=sender_name or sender_id)
msg_type = MessageType.TEXT
if media_types:
if any(mt.startswith("audio/") for mt in media_types):
msg_type = MessageType.VOICE
elif any(mt.startswith("image/") for mt in media_types):
msg_type = MessageType.PHOTO
else:
# Non-image/non-audio files are documents so run.py's
# document-context injection surfaces the file to the agent.
else: # other files are DOCUMENT so run.py's document-context injection surfaces them
msg_type = MessageType.DOCUMENT
ts_str = meta.get("itemTs") or meta.get("createdAt", "")
try:
timestamp = (datetime.fromisoformat(ts_str.replace("Z", "+00:00")) if ts_str
else datetime.now(tz=timezone.utc))
timestamp = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) if ts_str else datetime.now(tz=timezone.utc)
except (ValueError, AttributeError):
timestamp = datetime.now(tz=timezone.utc)
msg_event = MessageEvent(
source=source, text=text or "", message_type=msg_type, media_urls=media_urls,
media_types=media_types, timestamp=timestamp, raw_message=chat_item,
)
logger.debug("SimpleX: message from %s in %s: %s",
_redact_id(sender_id), chat_id[:20], (text or "")[:50])
# Batch rapid-fire text so the agent sees one combined message.
if msg_type == MessageType.TEXT and text:
media_types=media_types, timestamp=timestamp, raw_message=chat_item)
logger.debug("SimpleX: message from %s in %s: %s", _redact_id(sender_id), chat_id[:20], (text or "")[:50])
if msg_type == MessageType.TEXT and text: # batch rapid-fire text into one combined message
self._enqueue_text_event(msg_event)
else:
await self.handle_message(msg_event)
# ------------------------------------------------------------------
# Text message batching (enqueue lives on BasePlatformAdapter)
# ------------------------------------------------------------------
# Text batching: enqueue lives on BasePlatformAdapter.
def _text_batch_key(self, event: MessageEvent) -> str:
"""Session-scoped key for text message batching."""
return f"{event.source.platform.value}:{event.source.chat_id}"
async def _flush_text_batch(self, key: str) -> None:
@@ -529,26 +413,14 @@ class SimplexAdapter(BasePlatformAdapter):
if self._pending_text_batch_tasks.get(key) is current_task:
self._pending_text_batch_tasks.pop(key, None)
# ------------------------------------------------------------------
# Command interface
# ------------------------------------------------------------------
def _make_corr_id(self) -> str:
"""Mint a correlation ID and remember it for echo-filtering.
``_pending_corr_ids`` is bounded: past ``_max_pending_corr`` the
overflow is evicted in a single sweep.
"""
"""Mint a correlation ID and remember it for echo-filtering; the set is bounded by
``_max_pending_corr`` (overflow evicted in one sweep)."""
self._corr_counter += 1
corr_id = f"{_CORR_PREFIX}{self._corr_counter}-{int(time.time() * 1000)}"
self._pending_corr_ids.add(corr_id)
if len(self._pending_corr_ids) > self._max_pending_corr:
overflow = len(self._pending_corr_ids) - self._max_pending_corr
for _ in range(overflow):
try:
self._pending_corr_ids.pop()
except KeyError:
break
for _ in range(len(self._pending_corr_ids) - self._max_pending_corr):
self._pending_corr_ids.pop()
return corr_id
async def _send_ws(self, payload: dict) -> None:
@@ -570,20 +442,17 @@ class SimplexAdapter(BasePlatformAdapter):
return None
corr_id = self._make_corr_id()
payload = json.dumps({"corrId": corr_id, "cmd": command})
fut: asyncio.Future = asyncio.get_event_loop().create_future()
self._pending_responses[corr_id] = fut
try:
await ws.send(payload)
await ws.send(json.dumps({"corrId": corr_id, "cmd": command}))
return await asyncio.wait_for(fut, timeout=timeout)
except asyncio.TimeoutError:
logger.warning("SimpleX: command timed out: %s", command[:50])
self._pending_responses.pop(corr_id, None)
return None
except Exception as e:
logger.warning("SimpleX: command failed: %s — %s", command[:50], e)
self._pending_responses.pop(corr_id, None)
return None
self._pending_responses.pop(corr_id, None)
return None
async def _send_fire_and_forget(self, command: str) -> None:
"""Send a command the daemon never replies to with a corrId (e.g. ``/freceive``)."""
@@ -592,113 +461,67 @@ class SimplexAdapter(BasePlatformAdapter):
async def _send_items(self, chat_id: str, items: list, error: str) -> SendResult:
"""Send a structured ``/_send`` payload and await the reply."""
result = await self._send_command(_send_cmd(chat_id, items))
if result is not None:
return SendResult(success=True)
return SendResult(success=False, error=error)
# ------------------------------------------------------------------
# Outbound — text
# ------------------------------------------------------------------
return SendResult(success=True) if result is not None else SendResult(success=False, error=error)
async def send(
self,
chat_id: str,
content: str,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Send a text message.
``MEDIA:<path>`` tags (embedded by TTS / audio tools) are stripped from
the body and sent as native voice notes or documents. The text send is
fire-and-forget at the WebSocket level: the daemon doesn't always return
a corrId reply for chat commands, and waiting would serialise all
outbound traffic behind a 30-second timeout.
"""
_voice_exts = {".ogg", ".mp3", ".wav", ".m4a", ".opus"}
"""Send text; ``MEDIA:<path>`` tags (TTS / audio tools) are stripped and sent as native voice
notes or documents. The text send is fire-and-forget: the daemon doesn't always return a corrId
reply for chat commands, and waiting would serialise all outbound traffic behind a 30s timeout."""
media_paths = re.findall(r"MEDIA:(\S+)", content)
if media_paths:
content = re.sub(r"MEDIA:\S+", "", content).strip()
if content:
corr_id = self._make_corr_id()
cmd_str = _send_cmd(chat_id, [{"msgContent": {"type": "text", "text": content}}])
await self._send_ws({"corrId": corr_id, "cmd": cmd_str})
await self._send_ws({"corrId": self._make_corr_id(), "cmd": cmd_str})
for path in media_paths:
if os.path.splitext(path)[1].lower() in _voice_exts:
if os.path.splitext(path)[1].lower() in _VOICE_TAG_EXTS:
media_result = await self.send_voice(chat_id, path)
else:
media_result = await self.send_document(chat_id, path)
if not media_result.success:
return media_result
return SendResult(success=True)
# ------------------------------------------------------------------
# Channel directory enumeration
# ------------------------------------------------------------------
async def list_channels(self) -> Optional[List[Dict[str, Any]]]:
"""Enumerate contacts and allowed groups for the channel directory.
Returns ``None`` (not ``[]``) when the WebSocket is down or the daemon
is unresponsive so the directory falls back to session-history
discovery instead of wiping known targets. Entry ``id`` values match
the adapter's send targets: display name for DMs, ``group:<id>`` for groups.
Returns ``None`` (not ``[]``) when the WebSocket is down or the daemon is unresponsive so
the directory falls back to session-history discovery instead of wiping known targets.
Entry ``id`` values match the send targets: display name for DMs, ``group:<id>`` for groups.
"""
if not self._ws:
return None
channels: List[Dict[str, Any]] = []
resp = await self._send_command("/contacts", timeout=10.0)
if resp is None:
return None
channels: List[Dict[str, Any]] = []
for contact in resp.get("contacts") or []:
if not isinstance(contact, dict):
continue
contact_id = contact.get("contactId")
name = (contact.get("localDisplayName", "")
or (contact.get("profile", {}) or {}).get("displayName", ""))
name = _display_name(contact, "profile")
if contact_id is None and not name:
continue
# Display name is what the DM send path addresses; fall back to contactId.
channels.append(
{"id": str(name or contact_id), "name": str(name or contact_id), "type": "dm"})
channels.append({"id": str(name or contact_id), "name": str(name or contact_id), "type": "dm"})
resp = await self._send_command("/groups", timeout=10.0)
if resp is not None:
for group in resp.get("groups") or []:
# Each group is either a groupInfo dict or a [groupInfo, groupSummary] pair.
if isinstance(group, list) and group:
group = group[0]
if not isinstance(group, dict):
continue
group_id = group.get("groupId")
if group_id is None:
continue
name = (
group.get("localDisplayName", "")
or (group.get("groupProfile", {}) or {}).get("displayName", "")
or str(group_id)
)
channels.append({"id": f"group:{group_id}", "name": str(name), "type": "group"})
for group in (resp.get("groups") or []) if resp is not None else []:
if isinstance(group, list) and group: # groupInfo dict or [groupInfo, groupSummary] pair
group = group[0]
if not isinstance(group, dict) or group.get("groupId") is None:
continue
group_id = group["groupId"]
name = _display_name(group, "groupProfile") or str(group_id)
channels.append({"id": f"group:{group_id}", "name": str(name), "type": "group"})
return channels
# ------------------------------------------------------------------
# Outbound — media
# ------------------------------------------------------------------
@staticmethod
def _prepare_image(file_path: str) -> tuple[str, str]:
"""Ensure *file_path* is PNG/JPEG and return ``(png_path, thumb_data_uri)``.
SimpleX clients can't display WebP etc. inline, so convert to PNG when
needed and build a 128px JPEG thumbnail for the ``image`` field (inline
preview). Uses Pillow when available, else ImageMagick ``convert``.
"""
"""Ensure *file_path* is PNG/JPEG and return ``(png_path, thumb_data_uri)``. SimpleX clients
can't show WebP etc. inline, so convert to PNG when needed and build a 128px JPEG thumbnail for
the ``image`` field. Uses Pillow when available, else ImageMagick ``convert``."""
import subprocess
import tempfile
@@ -706,7 +529,6 @@ class SimplexAdapter(BasePlatformAdapter):
png_path = file_path
thumb_uri = ""
needs_png = p.suffix.lower() not in (".png", ".jpg", ".jpeg")
try:
from PIL import Image
import io
@@ -728,10 +550,8 @@ class SimplexAdapter(BasePlatformAdapter):
check=True, capture_output=True, timeout=30, stdin=subprocess.DEVNULL)
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
tmp_path = tmp.name
subprocess.run(
["convert", file_path, "-resize", "128x128", "-quality", "70", tmp_path],
check=True, capture_output=True, timeout=30,
)
subprocess.run(["convert", file_path, "-resize", "128x128", "-quality", "70", tmp_path],
check=True, capture_output=True, timeout=30)
with open(tmp_path, "rb") as f:
thumb_uri = "data:image/jpg;base64," + base64.b64encode(f.read()).decode()
os.remove(tmp_path)
@@ -741,49 +561,38 @@ class SimplexAdapter(BasePlatformAdapter):
return png_path, thumb_uri
async def send_image(
self, chat_id: str, image_url: str, caption: Optional[str] = None, **kwargs
) -> SendResult:
self, chat_id: str, image_url: str, caption: Optional[str] = None, **kwargs) -> SendResult:
"""Send an image. Supports ``file://`` URLs and ``http(s)://`` URLs."""
from urllib.parse import unquote
if image_url.startswith("file://"):
file_path = unquote(image_url[7:])
else:
try:
from gateway.platforms.base import cache_image_from_url
file_path = await cache_image_from_url(image_url)
except Exception as e:
logger.warning("SimpleX: failed to download image: %s", e)
return SendResult(success=False, error=str(e))
if not file_path or not Path(file_path).exists():
return SendResult(success=False, error="Image file not found")
png_path, thumb_uri = self._prepare_image(file_path)
# /_send addresses by numeric ID; /f only accepts display names.
item = {"filePath": png_path,
"msgContent": {"type": "image", "image": thumb_uri, "text": caption or ""}}
item = {"filePath": png_path, "msgContent": {"type": "image", "image": thumb_uri, "text": caption or ""}}
return await self._send_items(chat_id, [item], "Failed to send image")
async def send_image_file(
self, chat_id: str, image_path: str, caption: Optional[str] = None,
reply_to: Optional[str] = None, **kwargs,
) -> SendResult:
reply_to: Optional[str] = None, **kwargs) -> SendResult:
"""Send a local image file via SimpleX."""
return await self.send_image(chat_id, f"file://{image_path}", caption=caption, **kwargs)
async def send_video(
self, chat_id: str, video_path: str, caption: Optional[str] = None,
reply_to: Optional[str] = None, **kwargs,
) -> SendResult:
reply_to: Optional[str] = None, **kwargs) -> SendResult:
"""Send a video file via SimpleX (as a file attachment)."""
return await self.send_document(chat_id, video_path, caption=caption)
async def send_document(
self, chat_id: str, file_path: str, caption: Optional[str] = None,
filename: Optional[str] = None, **kwargs,
) -> SendResult:
filename: Optional[str] = None, **kwargs) -> SendResult:
"""Send a document/file attachment."""
if not Path(file_path).exists():
return SendResult(success=False, error="File not found")
@@ -792,15 +601,12 @@ class SimplexAdapter(BasePlatformAdapter):
async def send_voice(
self, chat_id: str, audio_path: str, caption: Optional[str] = None,
reply_to: Optional[str] = None, duration: int = 0, **kwargs,
) -> SendResult:
reply_to: Optional[str] = None, duration: int = 0, **kwargs) -> SendResult:
"""Send an audio file as an inline SimpleX voice note (``msgContent.type == "voice"``)."""
if not Path(audio_path).exists():
return SendResult(success=False, error="Voice file not found")
item = {
"msgContent": {"type": "voice", "text": caption or "", "duration": duration},
"fileSource": {"filePath": audio_path},
}
item = {"msgContent": {"type": "voice", "text": caption or "", "duration": duration},
"fileSource": {"filePath": audio_path}}
return await self._send_items(chat_id, [item], "Failed to send voice message")
async def send_typing(self, chat_id: str, metadata=None) -> None:
@@ -813,10 +619,6 @@ class SimplexAdapter(BasePlatformAdapter):
return {"chat_id": chat_id, "type": "dm", "name": chat_id}
# ---------------------------------------------------------------------------
# Plugin entry-point hooks
# ---------------------------------------------------------------------------
def check_requirements() -> bool:
"""Plugin gate: require SIMPLEX_WS_URL AND the websockets package."""
if not _get_scoped_secret("SIMPLEX_WS_URL"):
@@ -840,70 +642,48 @@ def is_connected(config) -> bool:
def _env_enablement() -> Optional[dict]:
"""Seed ``PlatformConfig.extra`` from env vars during gateway config load.
Runs BEFORE adapter construction so ``gateway status`` reflects env-only
configuration. Returns ``None`` when SimpleX isn't minimally configured.
``home_channel`` is turned into a ``HomeChannel`` by the core hook.
"""
"""Seed ``PlatformConfig.extra`` from env BEFORE adapter construction so ``gateway status``
reflects env-only setups. ``None`` when not minimally configured; ``home_channel`` becomes
a ``HomeChannel`` via the core hook."""
ws_url = _get_scoped_secret("SIMPLEX_WS_URL", "").strip()
if not ws_url:
return None
seed: dict = {"ws_url": ws_url}
auto_accept = _get_scoped_secret("SIMPLEX_AUTO_ACCEPT", "").strip().lower()
if auto_accept:
seed["auto_accept"] = auto_accept not in {"0", "false", "no"}
group_allowed = _get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "").strip()
if group_allowed:
seed["group_allowed"] = group_allowed
home = _get_scoped_secret("SIMPLEX_HOME_CHANNEL", "").strip()
if home:
seed["home_channel"] = {
"chat_id": home,
"name": _get_scoped_secret("SIMPLEX_HOME_CHANNEL_NAME", "").strip() or home,
}
seed["home_channel"] = {"chat_id": home, "name": _get_scoped_secret("SIMPLEX_HOME_CHANNEL_NAME", "").strip() or home}
return seed
async def _standalone_send(
pconfig,
chat_id: str,
message: str,
*,
thread_id: Optional[str] = None,
media_files: Optional[List[str]] = None,
force_document: bool = False,
pconfig, chat_id: str, message: str, *,
thread_id: Optional[str] = None, media_files: Optional[List[str]] = None, force_document: bool = False,
) -> Dict[str, Any]:
"""Open an ephemeral WebSocket to the daemon, send, and close.
Used by ``tools/send_message_tool`` when the gateway runner is not in this
process (e.g. ``hermes cron``). ``thread_id``/``force_document`` are
signature parity only; ``media_files`` is accepted but only the text body
is delivered — SimpleX file transfers need the daemon's filesystem-backed
flow, which an ephemeral connection cannot drive safely.
"""
"""Ephemeral WebSocket send for ``tools/send_message_tool`` when the gateway runner is not in
this process (``hermes cron``). ``thread_id``/``force_document`` are signature parity only;
``media_files`` is accepted but only the text body is delivered — SimpleX file transfers need
the daemon's filesystem-backed flow, which an ephemeral connection cannot drive safely."""
try:
import websockets as _wsclient
except ImportError:
return {"error": "websockets not installed. Run: pip install websockets"}
extra = getattr(pconfig, "extra", {}) or {}
ws_url = _get_scoped_secret("SIMPLEX_WS_URL") or extra.get("ws_url", "ws://127.0.0.1:5225")
if not ws_url:
return {"error": "SimpleX standalone send: SIMPLEX_WS_URL is required"}
try:
payload = {
"corrId": f"{_CORR_PREFIX}snd-{int(time.time() * 1000)}",
"cmd": _send_cmd(chat_id, [{"msgContent": {"type": "text", "text": message}}]),
}
"cmd": _send_cmd(chat_id, [{"msgContent": {"type": "text", "text": message}}])}
async with _wsclient.connect(ws_url, open_timeout=10, close_timeout=5) as ws:
await ws.send(json.dumps(payload))
# Give the daemon a moment to process the command before closing.
await asyncio.sleep(0.5)
await asyncio.sleep(0.5) # let the daemon process the command before closing
return {"success": True, "platform": "simplex", "chat_id": chat_id}
except Exception as e:
return {"error": f"SimpleX send failed: {e}"}
@@ -911,14 +691,10 @@ async def _standalone_send(
def interactive_setup() -> None:
"""Minimal stdin wizard for ``hermes setup gateway`` → SimpleX; writes ``~/.hermes/.env``."""
print()
print("SimpleX Chat setup")
print("------------------")
print("Requirements:")
print(" 1. simplex-chat daemon running (e.g. `simplex-chat -p 5225`).")
print(" 2. Python package `websockets` installed (`pip install websockets`).")
print()
print(
"\nSimpleX Chat setup\n------------------\nRequirements:\n"
" 1. simplex-chat daemon running (e.g. `simplex-chat -p 5225`).\n"
" 2. Python package `websockets` installed (`pip install websockets`).\n")
try:
from hermes_cli.config import get_env_value, save_env_value
except ImportError:
@@ -937,12 +713,9 @@ def interactive_setup() -> None:
save_env_value(var, value)
_prompt("SIMPLEX_WS_URL", "Daemon WebSocket URL (default ws://127.0.0.1:5225)")
_prompt("SIMPLEX_ALLOWED_USERS",
"Allowed contactIds or display names (comma-separated; blank=skip)")
_prompt("SIMPLEX_GROUP_ALLOWED",
"Allowed group IDs (comma-separated, or '*' for any; blank=disable groups)")
_prompt("SIMPLEX_AUTO_ACCEPT",
"Auto-accept incoming contact requests? (true/false, default true)")
_prompt("SIMPLEX_ALLOWED_USERS", "Allowed contactIds or display names (comma-separated; blank=skip)")
_prompt("SIMPLEX_GROUP_ALLOWED", "Allowed group IDs (comma-separated, or '*' for any; blank=disable groups)")
_prompt("SIMPLEX_AUTO_ACCEPT", "Auto-accept incoming contact requests? (true/false, default true)")
_prompt("SIMPLEX_HOME_CHANNEL", "Home channel contact/group ID (or empty)")
print("Done. Make sure the simplex-chat daemon is running before starting the gateway.")
@@ -977,6 +750,4 @@ def register(ctx) -> None:
"hard message length limit, but keep responses conversational. "
"You can attach native images, voice notes, and arbitrary "
"files; the adapter handles MEDIA:<path> tags by sending them "
"as inline voice notes (audio extensions) or documents."
),
)
"as inline voice notes (audio extensions) or documents."))
plugins/platforms/teams/adapter.py
+175 -304
View File
@@ -1,21 +1,10 @@
"""
Microsoft Teams platform adapter: microsoft-teams-apps SDK for auth/activity
processing, an aiohttp webhook server for inbound, ``App.send()`` for proactive sends.
"""Microsoft Teams adapter: microsoft-teams-apps SDK for auth/activity processing, an aiohttp
webhook server for inbound, ``App.send()`` for proactive sends.
Requires:
the ``teams`` extra (auto-installed by the gateway on first start, or
manually: ``<hermes-venv>/bin/pip install microsoft-teams-apps aiohttp``)
TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID env vars
Configuration in config.yaml:
platforms:
teams:
enabled: true
extra:
client_id: "your-client-id" # or TEAMS_CLIENT_ID env var
client_secret: "your-secret" # or TEAMS_CLIENT_SECRET env var
tenant_id: "your-tenant-id" # or TEAMS_TENANT_ID env var
port: 3978 # or TEAMS_PORT env var
Requires the ``teams`` extra (auto-installed by the gateway on first start, or
``<hermes-venv>/bin/pip install microsoft-teams-apps aiohttp``) and credentials via env
(TEAMS_CLIENT_ID / TEAMS_CLIENT_SECRET / TEAMS_TENANT_ID, optional TEAMS_PORT) or
``platforms.teams.extra`` in config.yaml (``client_id`` / ``client_secret`` / ``tenant_id`` / ``port``).
"""
from __future__ import annotations
@@ -33,7 +22,6 @@ from urllib.parse import urlparse
try:
from aiohttp import web
AIOHTTP_AVAILABLE = True
except ImportError:
AIOHTTP_AVAILABLE = False
@@ -41,20 +29,16 @@ except ImportError:
def _probe_teams_sdk_available() -> bool:
"""True when ``microsoft_teams.apps`` is on sys.path, without importing it.
Import is deferred: ``microsoft_teams.apps.app`` loads a cwd ``.env`` at import
(#62935); ``check_teams_requirements()`` binds symbols behind a dotenv no-op.
Sibling packages share the ``microsoft_teams`` namespace, so probe the parent
first — ``find_spec("microsoft_teams.apps")`` raises on 3.11+ if it is absent.
"""
"""True when ``microsoft_teams.apps`` is on sys.path, without importing it: the SDK loads a cwd
``.env`` at import, so ``check_teams_requirements()`` binds symbols behind a dotenv no-op.
Sibling packages share the namespace, so probe the parent first — ``find_spec`` of the child
raises on 3.11+ if the parent is absent."""
try:
if importlib.util.find_spec("microsoft_teams") is None:
return False
return importlib.util.find_spec("microsoft_teams.apps") is not None
except (ValueError, ModuleNotFoundError, ImportError):
# Test stubs may inject a module without ``__spec__``.
return "microsoft_teams.apps" in sys.modules
return "microsoft_teams.apps" in sys.modules # test stubs may lack ``__spec__``
TEAMS_SDK_AVAILABLE = _probe_teams_sdk_available()
@@ -68,13 +52,7 @@ HttpMethod = str # type: ignore[assignment,misc]
from gateway.config import Platform, PlatformConfig
from gateway.platforms.helpers import MessageDeduplicator
from gateway.platforms.base import (
gateway_trust_env,
BasePlatformAdapter,
MessageEvent,
MessageType,
SendResult,
cache_image_from_url,
cache_media_bytes,
gateway_trust_env, BasePlatformAdapter, MessageEvent, MessageType, SendResult, cache_image_from_url, cache_media_bytes,
)
from gateway.platforms._shared import coerce_port, get_scoped_secret as _get_scoped_secret
from plugins.platforms.teams.summary_writer import TeamsSummaryWriter # noqa: F401 — re-exported for teams_pipeline
@@ -82,28 +60,19 @@ from plugins.platforms.teams.summary_writer import TeamsSummaryWriter # noqa: F
logger = logging.getLogger(__name__)
_DEFAULT_PORT = 3978
# Bot Framework activities are JSON well under 1 MiB; bound oversized/chunked bodies.
_MAX_BODY_BYTES = 1_048_576
# ``None`` host → aiohttp binds IPv4 + IPv6 (a hardcoded "0.0.0.0" was unreachable
# on IPv6-only networks such as Fly.io 6PN). Pin via TEAMS_HOST or extra.host.
_MAX_BODY_BYTES = 1_048_576 # Bot Framework activities are JSON well under 1 MiB
# ``None`` host → aiohttp binds IPv4 + IPv6 ("0.0.0.0" was unreachable on IPv6-only
# networks such as Fly.io 6PN). Pin via TEAMS_HOST or extra.host.
_DEFAULT_HOST = None
_WEBHOOK_PATH = "/api/messages"
# Bot Framework default service URL. Regional/government tenants may need a
# different host (e.g. ``https://smba.infra.gov.teams.microsoft.us/``) via
# ``TEAMS_SERVICE_URL`` or ``extra['service_url']``.
# Regional/government tenants override via ``TEAMS_SERVICE_URL`` / ``extra['service_url']``.
_DEFAULT_TEAMS_SERVICE_URL = "https://smba.trafficmanager.net/teams/"
# Bot Framework hosts that may receive a freshly minted bearer token (blocks SSRF /
# token exfiltration via a tampered env var). Exact match only: any Azure customer can
# register ``<name>.trafficmanager.net``. New regions are allowlist additions.
# Hosts that may receive a freshly minted bearer token (blocks SSRF / token exfiltration via a
# tampered env var). Exact match only: any Azure customer can register ``<name>.trafficmanager.net``.
_ALLOWED_TEAMS_SERVICE_HOSTS = frozenset({"smba.trafficmanager.net", "smba.infra.gov.teams.microsoft.us"})
# Conservative Bot Framework conversation-ID charset (digits, colons, hyphens,
# dots, '@', ``thread.skype`` / ``thread.tacv2`` suffixes) so a hostile value
# cannot path-traverse out of ``/v3/conversations/<id>/activities``.
# Conservative conversation-ID charset (``thread.skype`` / ``thread.tacv2`` suffixes included) so a
# hostile value cannot path-traverse out of ``/v3/conversations/<id>/activities``.
_TEAMS_CONV_ID_RE = re.compile(r"^[A-Za-z0-9:@\-_.]+$")
_BF_TOKEN_SCOPE = "https://api.botframework.com/.default"
@@ -111,12 +80,7 @@ def _bf_token_request(tenant_id: str, client_id: str, client_secret: str) -> tup
"""(token URL, client-credentials form) for a Bot Framework bearer token."""
return (
f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
{
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": _BF_TOKEN_SCOPE,
},
{"grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, "scope": _BF_TOKEN_SCOPE},
)
@@ -124,9 +88,7 @@ def _is_allowed_https_host(url: str, *, check_port: bool = False) -> bool:
"""https + host in ``_ALLOWED_TEAMS_SERVICE_HOSTS`` (+ default port when asked)."""
try:
parsed = urlparse(url)
if parsed.scheme != "https":
return False
if check_port and parsed.port not in (None, 443):
if parsed.scheme != "https" or (check_port and parsed.port not in (None, 443)):
return False
return parsed.hostname in _ALLOWED_TEAMS_SERVICE_HOSTS
except Exception:
@@ -146,19 +108,15 @@ def _validate_teams_service_url(raw: str) -> Optional[str]:
class _AiohttpBridgeAdapter:
"""HttpServerAdapter bridging SDK route registrations into our aiohttp app.
Without it ``App()`` unconditionally imports fastapi/uvicorn and allocates a ``FastAPI()``.
"""
"""HttpServerAdapter bridging SDK route registrations into our aiohttp app; without it
``App()`` unconditionally imports fastapi/uvicorn and allocates a ``FastAPI()``."""
def __init__(self, aiohttp_app: "web.Application"):
self._aiohttp_app = aiohttp_app
def register_route(self, method: "HttpMethod", path: str, handler: "HttpRouteHandler") -> None:
async def _aiohttp_handler(request: "web.Request") -> "web.Response":
body = await request.json()
headers = dict(request.headers)
result: "HttpResponse" = await handler(HttpRequest(body=body, headers=headers))
result: "HttpResponse" = await handler(HttpRequest(body=await request.json(), headers=dict(request.headers)))
status = result.get("status", 200)
resp_body = result.get("body")
if resp_body is not None:
@@ -188,8 +146,7 @@ def _credentials(config) -> tuple[str, str, str]:
return (
os.getenv("TEAMS_CLIENT_ID") or extra.get("client_id", ""),
_get_scoped_secret("TEAMS_CLIENT_SECRET") or extra.get("client_secret", ""),
os.getenv("TEAMS_TENANT_ID") or extra.get("tenant_id", ""),
)
os.getenv("TEAMS_TENANT_ID") or extra.get("tenant_id", ""))
def validate_config(config) -> bool:
@@ -203,11 +160,9 @@ def is_connected(config) -> bool:
def _env_enablement() -> dict | None:
"""Seed ``PlatformConfig.extra`` from env vars before adapter construction.
Lets ``gateway status`` reflect env-only setups without the SDK. ``None`` when
not minimally configured; ``home_channel`` becomes a ``HomeChannel`` via the core hook.
"""
"""Seed ``PlatformConfig.extra`` from env before adapter construction so ``gateway status`` reflects
env-only setups without the SDK. ``None`` when not minimally configured; ``home_channel`` becomes a
``HomeChannel`` via the core hook."""
client_id = os.getenv("TEAMS_CLIENT_ID", "").strip()
client_secret = _get_scoped_secret("TEAMS_CLIENT_SECRET", "").strip()
tenant_id = os.getenv("TEAMS_TENANT_ID", "").strip()
@@ -230,18 +185,14 @@ async def _standalone_send(
pconfig, chat_id: str, message: str, *,
thread_id: Optional[str] = None, media_files: Optional[list] = None, force_document: bool = False,
) -> Dict[str, Any]:
"""Acquire a Bot Framework bearer token and POST a single message activity.
Used by ``send_message_tool._send_via_adapter`` when the gateway runner is
not in this process (``hermes cron``). ``TEAMS_SERVICE_URL`` is allowlisted
and ``chat_id`` charset-checked (SSRF/path traversal). ``media_files`` /
``force_document`` are signature parity only — this path is text-only.
"""
"""Acquire a Bot Framework bearer token and POST a single message activity; used by
``send_message_tool._send_via_adapter`` when the gateway runner is not in this process
(``hermes cron``). ``TEAMS_SERVICE_URL`` is allowlisted and ``chat_id`` charset-checked
(SSRF/path traversal). ``media_files`` / ``force_document`` are signature parity only — text-only."""
extra = getattr(pconfig, "extra", {}) or {}
client_id, client_secret, tenant_id = _credentials(pconfig)
if not (client_id and client_secret and tenant_id):
return {"error": "Teams standalone send: TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID are all required"}
raw_service_url = os.getenv("TEAMS_SERVICE_URL") or extra.get("service_url", "") or _DEFAULT_TEAMS_SERVICE_URL
service_url = _validate_teams_service_url(raw_service_url)
if service_url is None:
@@ -253,13 +204,10 @@ async def _standalone_send(
return {"error": "Teams standalone send: chat_id contains characters outside the Bot Framework conversation ID set"}
if not _TEAMS_CONV_ID_RE.match(tenant_id):
return {"error": "Teams standalone send: TEAMS_TENANT_ID contains characters outside the expected set"}
token_url, token_form = _bf_token_request(tenant_id, client_id, client_secret)
activities_url = f"{service_url}v3/conversations/{chat_id}/activities"
if not AIOHTTP_AVAILABLE:
return {"error": "Teams standalone send: aiohttp not installed"}
try:
import aiohttp as _aiohttp
@@ -267,9 +215,7 @@ async def _standalone_send(
per_request_timeout = _aiohttp.ClientTimeout(total=15.0)
async with _aiohttp.ClientSession(trust_env=gateway_trust_env()) as session:
async with session.post(
token_url,
data=token_form,
headers={"Content-Type": "application/x-www-form-urlencoded"},
token_url, data=token_form, headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=per_request_timeout,
) as token_resp:
if token_resp.status >= 400:
@@ -279,11 +225,8 @@ async def _standalone_send(
access_token = token_payload.get("access_token")
if not access_token:
return {"error": "Teams standalone send: token response missing access_token"}
activity = {"type": "message", "text": message, "textFormat": "markdown"}
async with session.post(
activities_url,
json=activity,
activities_url, json={"type": "message", "text": message, "textFormat": "markdown"},
headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"},
timeout=per_request_timeout,
) as send_resp:
@@ -309,17 +252,13 @@ _SDK_IMPORTS = {
"microsoft_teams.api.models.adaptive_card": ("AdaptiveCardActionCardResponse", "AdaptiveCardActionMessageResponse"),
"microsoft_teams.api.models.invoke_response": ("InvokeResponse", "AdaptiveCardInvokeResponse"),
"microsoft_teams.apps.http.adapter": ("HttpMethod", "HttpRequest", "HttpResponse", "HttpRouteHandler"),
"microsoft_teams.cards": ("AdaptiveCard", "ExecuteAction", "TextBlock"),
}
"microsoft_teams.cards": ("AdaptiveCard", "ExecuteAction", "TextBlock")}
@contextmanager
def _suppress_third_party_dotenv() -> Iterator[None]:
"""No-op ``dotenv.load_dotenv`` while importing the Teams SDK (#62935).
``microsoft_teams.apps.app`` loads a cwd-discovered ``.env`` at import,
mutating process-global ``os.environ``. Hermes owns dotenv loading.
"""
"""No-op ``dotenv.load_dotenv`` while importing the Teams SDK: ``microsoft_teams.apps.app`` loads a
cwd-discovered ``.env`` at import, mutating process-global ``os.environ``. Hermes owns dotenv loading."""
try:
import dotenv as _dotenv
except ImportError:
@@ -336,12 +275,9 @@ def _suppress_third_party_dotenv() -> Iterator[None]:
def check_teams_requirements() -> bool:
"""ACTIVE lazy-installer (registry ``ensure_deps_fn``): install the SDK on first use.
Rebinds all module-level SDK globals on success; True once the SDK and
aiohttp are importable. Gate on ``App is not None`` — ``TEAMS_SDK_AVAILABLE``
is only a find_spec probe and can be True before any import ran.
"""
"""ACTIVE lazy-installer (registry ``ensure_deps_fn``): install the SDK on first use and rebind
the module-level SDK globals. Gate on ``App is not None`` — ``TEAMS_SDK_AVAILABLE`` is only a
find_spec probe and can be True before any import ran."""
if App is not None and AIOHTTP_AVAILABLE:
return True
@@ -361,7 +297,6 @@ def check_teams_requirements() -> bool:
return bindings
from tools.lazy_deps import ensure_and_bind
return ensure_and_bind("platform.teams", _import, globals(), prompt=False)
@@ -370,14 +305,28 @@ _CHAT_TYPES = {"personal": "dm", "groupChat": "group", "channel": "channel"}
# injection gates strictly on MessageType.DOCUMENT (same precedence as Email/Signal).
_MEDIA_KIND_PRECEDENCE = (
("document", MessageType.DOCUMENT), ("image", MessageType.PHOTO),
("video", MessageType.VIDEO), ("audio", MessageType.AUDIO),
)
("video", MessageType.VIDEO), ("audio", MessageType.AUDIO))
_APPROVAL_CHOICES = {"approve_once": "once", "approve_session": "session", "approve_always": "always", "deny": "deny"}
_APPROVAL_LABELS = {
"once": "✅ Allowed (once)", "session": "✅ Allowed (session)", "always": "✅ Always allowed", "deny": "❌ Denied",
}
def _truncate(text: str, limit: int) -> str:
return text[:limit] + "..." if len(text) > limit else text
def _approval_body(cmd: str, desc: str, *, always: bool = False) -> list:
"""Adaptive Card body blocks for an approval prompt; unless ``always``, empty ``cmd``/``desc`` omit their blocks."""
body = []
if cmd or always:
body.append(TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"))
body.append(TextBlock(text=f"```\n{cmd}\n```", wrap=True))
if desc or always:
body.append(TextBlock(text=f"Reason: {desc}", wrap=True, isSubtle=True))
return body
class TeamsAdapter(BasePlatformAdapter):
"""Microsoft Teams adapter using the microsoft-teams-apps SDK."""
@@ -390,24 +339,21 @@ class TeamsAdapter(BasePlatformAdapter):
self._client_id = extra.get("client_id") or os.getenv("TEAMS_CLIENT_ID", "")
self._client_secret = extra.get("client_secret") or _get_scoped_secret("TEAMS_CLIENT_SECRET", "")
self._tenant_id = extra.get("tenant_id") or os.getenv("TEAMS_TENANT_ID", "")
# (token, expiry monotonic ts) for connector attachment auth; refreshed
# under _bf_token_lock so concurrent attachments can't stampede the STS.
# (token, expiry monotonic ts) for connector attachment auth; refreshed under
# _bf_token_lock so concurrent attachments can't stampede the STS.
self._bf_token_cache: Optional[tuple] = None
self._bf_token_lock: Optional[asyncio.Lock] = None
self._port = coerce_port(extra.get("port") or os.getenv("TEAMS_PORT", str(_DEFAULT_PORT)), _DEFAULT_PORT)
# Falsy host (unset/"") collapses to the dual-stack default (None).
_raw_host = extra.get("host") or os.getenv("TEAMS_HOST", "") or _DEFAULT_HOST
_raw_host = extra.get("host") or os.getenv("TEAMS_HOST", "") or _DEFAULT_HOST # falsy → dual-stack None
self._host: Optional[str] = str(_raw_host) if _raw_host else None
self._app: Optional["App"] = None
self._runner: Optional["web.AppRunner"] = None
self._dedup = MessageDeduplicator(max_size=1000)
# chat_id → ConversationReference from incoming messages, so proactive
# cards use the right conversation type (personal/group/channel).
# chat_id → ConversationReference so proactive cards use the right conversation type.
self._conv_refs: Dict[str, Any] = {}
async def connect(self, *, is_reconnect: bool = False) -> bool:
# create_adapter() already ran the installer, but reconnect paths reach
# here too — re-run to bind SDK globals. Gate on App, not the find_spec flag.
# Reconnect paths reach here without create_adapter()'s installer — re-run to bind SDK globals.
check_teams_requirements()
pip = f"{sys.executable} -m pip install"
for failed, code, message in (
@@ -415,8 +361,7 @@ class TeamsAdapter(BasePlatformAdapter):
f"microsoft-teams-apps could not be installed. Run: {pip} microsoft-teams-apps"),
(not AIOHTTP_AVAILABLE, "MISSING_SDK", f"aiohttp not installed. Run: {pip} aiohttp"),
(not self._client_id or not self._client_secret or not self._tenant_id, "MISSING_CREDENTIALS",
"TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID are all required"),
):
"TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET, and TEAMS_TENANT_ID are all required")):
if failed:
self._set_fatal_error(code, message, retryable=False)
return False
@@ -425,18 +370,12 @@ class TeamsAdapter(BasePlatformAdapter):
# aiohttp app first — the bridge adapter wires SDK routes into it.
aiohttp_app = web.Application(client_max_size=_MAX_BODY_BYTES)
aiohttp_app.router.add_get("/health", lambda _: web.Response(text="ok"))
self._app = App(
client_id=self._client_id,
client_secret=self._client_secret,
tenant_id=self._tenant_id,
client_id=self._client_id, client_secret=self._client_secret, tenant_id=self._tenant_id,
http_server_adapter=_AiohttpBridgeAdapter(aiohttp_app),
client=ClientOptions(headers={"User-Agent": "Hermes"}),
)
# Handlers (ours, then plugin-registered on_* decorators) must be
# wired before initialize(), which registers POST /api/messages
# on aiohttp_app via the bridge's register_route().
client=ClientOptions(headers={"User-Agent": "Hermes"}))
# Handlers (ours, then plugin on_* decorators) must be wired before initialize(),
# which registers POST /api/messages on aiohttp_app via the bridge's register_route().
@self._app.on_message
async def _handle_message(ctx: ActivityContext[MessageActivity]):
await self._on_message(ctx)
@@ -449,20 +388,15 @@ class TeamsAdapter(BasePlatformAdapter):
self._wire_plugin_handlers(self._app)
await self._app.initialize()
self._runner = web.AppRunner(aiohttp_app)
await self._runner.setup()
site = web.TCPSite(self._runner, self._host, self._port)
await site.start()
self._running = True
self._mark_connected()
logger.info(
"[teams] Webhook server listening on %s:%d%s",
self._host or "* (all interfaces, IPv4+IPv6)",
self._port,
_WEBHOOK_PATH,
)
self._host or "* (all interfaces, IPv4+IPv6)", self._port, _WEBHOOK_PATH)
return True
except Exception as e:
self._set_fatal_error("CONNECT_FAILED", f"Teams connection failed: {e}", retryable=True)
@@ -479,11 +413,9 @@ class TeamsAdapter(BasePlatformAdapter):
logger.info("[teams] Disconnected")
async def _get_botframework_token(self) -> str:
"""Bot Framework bearer token (client credentials), cached until ~5 min before expiry.
Connector attachments are NOT pre-authenticated, unlike SharePoint downloadUrls.
Lock is created lazily: ``asyncio.Lock()`` in __init__ may bind the wrong loop.
"""
"""Bot Framework bearer token (client credentials), cached until ~5 min before expiry; connector
attachments are NOT pre-authenticated, unlike SharePoint downloadUrls. The lock is created lazily
because ``asyncio.Lock()`` in __init__ may bind the wrong loop."""
import time
import httpx
@@ -506,11 +438,8 @@ class TeamsAdapter(BasePlatformAdapter):
return token
async def _fetch_attachment_bytes(self, url: str, timeout: float = 30.0) -> bytes:
"""Download attachment bytes with SSRF protection.
Bot Framework connector URLs get the bot's bearer token; redirects and body
size go through the shared guards (as the cache_*_from_url helpers).
"""
"""Download attachment bytes with SSRF protection. Connector URLs get the bot's bearer token;
redirects and body size go through the shared guards (as the cache_*_from_url helpers)."""
from tools.url_safety import create_ssrf_safe_async_client, is_safe_url
from gateway.platforms.base import _ssrf_redirect_guard, _read_httpx_body_with_limit
@@ -539,16 +468,13 @@ class TeamsAdapter(BasePlatformAdapter):
msg_id = getattr(activity, "id", None)
if msg_id and self._dedup.is_duplicate(msg_id):
return
# Cache the conversation reference for proactive sends (approval cards, etc.)
conv_id = getattr(activity.conversation, "id", None)
if conv_id:
conv = activity.conversation
conv_id = getattr(conv, "id", None)
if conv_id: # cache the conversation reference for proactive sends (approval cards, etc.)
self._conv_refs[conv_id] = ctx.conversation_ref
text = activity.text if hasattr(activity, "text") and activity.text else ""
if "<at>" in text: # strip the <at>BotName</at> tags Teams prepends for @mentions
text = re.sub(r"<at>[^<]*</at>\s*", "", text).strip()
conv = activity.conversation
from_account = activity.from_
user_id = getattr(from_account, "aad_object_id", None) or getattr(from_account, "id", "")
source = self.build_source(
@@ -557,92 +483,79 @@ class TeamsAdapter(BasePlatformAdapter):
chat_type=_CHAT_TYPES.get(getattr(conv, "conversation_type", None) or "", "dm"),
user_id=str(user_id),
user_name=getattr(from_account, "name", None) or "",
guild_id=getattr(conv, "tenant_id", None) or self._tenant_id,
)
media_urls: list = []
media_types: list = []
media_kinds: list = []
def _add_cached(cached: Any, kind: Optional[str] = None) -> bool:
if not cached:
return False
media_urls.append(cached.path)
media_types.append(cached.media_type)
media_kinds.append(kind or cached.kind)
return True
guild_id=getattr(conv, "tenant_id", None) or self._tenant_id)
media: list = [] # (path, media_type, kind)
for att in getattr(activity, "attachments", None) or []:
content_url = getattr(att, "content_url", None)
content_type = (getattr(att, "content_type", None) or "").lower()
att_name = getattr(att, "name", None) or ""
# Skip non-file payloads: Teams mirrors the message body as a
# text/html attachment, and cards arrive as application/vnd.microsoft.card.*
if content_type in ("text/html", "text/plain") and not content_url:
continue
if content_type.startswith("application/vnd.microsoft.card"):
continue
if content_type == "application/vnd.microsoft.teams.file.download.info":
# Consent-free download: content carries a pre-authed SharePoint downloadUrl + file type.
content = getattr(att, "content", None)
if not isinstance(content, dict):
content = getattr(content, "__dict__", None) or {}
download_url = content.get("downloadUrl") or content.get("download_url")
file_type = (content.get("fileType") or content.get("file_type") or "").lstrip(".")
if not download_url:
continue
filename = att_name or (f"document.{file_type}" if file_type else "document")
try:
data = await self._fetch_attachment_bytes(download_url)
if not _add_cached(cache_media_bytes(data, filename=filename, mime_type="")):
logger.warning("[teams] Unsupported document type for attachment '%s', skipping", filename)
except Exception as e:
logger.warning("[teams] Failed to cache file attachment '%s': %s", filename, e)
continue
if content_url and content_type.startswith("image/"):
try:
if _is_botframework_attachment_url(content_url):
# Connector URL needs the bot's bearer token; the generic cache helper sends none.
data = await self._fetch_attachment_bytes(content_url)
ext = content_type.split("/")[-1].split(";")[0] or "png"
cached_m = cache_media_bytes(data, filename=att_name or f"image.{ext}", mime_type=content_type)
if not _add_cached(cached_m, kind="image"):
logger.warning(
"[teams] Bot Framework attachment '%s' returned data that failed image validation, skipping",
att_name or content_url,
)
else:
cached = await cache_image_from_url(content_url)
if cached:
media_urls.append(cached)
media_types.append(content_type)
media_kinds.append("image")
except Exception as e:
logger.warning("[teams] Failed to cache image attachment: %s", e)
continue
if content_url: # direct-URL non-image attachment (video/audio/document)
try:
data = await self._fetch_attachment_bytes(content_url)
_add_cached(cache_media_bytes(data, filename=att_name, mime_type=content_type))
except Exception as e:
logger.warning(
"[teams] Failed to cache attachment '%s' (%s): %s", att_name or content_url, content_type, e
)
cached = await self._cache_attachment(att)
if cached:
media.append(cached)
media_kinds = [kind for _, _, kind in media]
msg_type = next((t for kind, t in _MEDIA_KIND_PRECEDENCE if kind in media_kinds), MessageType.TEXT)
await self.handle_message(MessageEvent(
text=text, source=source, message_type=msg_type,
media_urls=media_urls, media_types=media_types, message_id=msg_id,
))
text=text, source=source, message_type=msg_type, message_id=msg_id,
media_urls=[path for path, _, _ in media], media_types=[mt for _, mt, _ in media]))
async def _cache_attachment(self, att: Any) -> Optional[tuple]:
"""Download + cache one inbound attachment → ``(path, media_type, kind)`` or ``None``."""
content_url = getattr(att, "content_url", None)
content_type = (getattr(att, "content_type", None) or "").lower()
att_name = getattr(att, "name", None) or ""
# Skip non-file payloads: Teams mirrors the message body as a text/html attachment,
# and cards arrive as application/vnd.microsoft.card.*
if content_type in ("text/html", "text/plain") and not content_url:
return None
if content_type.startswith("application/vnd.microsoft.card"):
return None
if content_type == "application/vnd.microsoft.teams.file.download.info":
# Consent-free download: content carries a pre-authed SharePoint downloadUrl + file type.
content = getattr(att, "content", None)
if not isinstance(content, dict):
content = getattr(content, "__dict__", None) or {}
download_url = content.get("downloadUrl") or content.get("download_url")
file_type = (content.get("fileType") or content.get("file_type") or "").lstrip(".")
if not download_url:
return None
filename = att_name or (f"document.{file_type}" if file_type else "document")
try:
data = await self._fetch_attachment_bytes(download_url)
cached = cache_media_bytes(data, filename=filename, mime_type="")
if not cached:
logger.warning("[teams] Unsupported document type for attachment '%s', skipping", filename)
return None
return cached.path, cached.media_type, cached.kind
except Exception as e:
logger.warning("[teams] Failed to cache file attachment '%s': %s", filename, e)
return None
if content_url and content_type.startswith("image/"):
try:
if _is_botframework_attachment_url(content_url):
# Connector URL needs the bot's bearer token; the generic cache helper sends none.
data = await self._fetch_attachment_bytes(content_url)
ext = content_type.split("/")[-1].split(";")[0] or "png"
cached = cache_media_bytes(data, filename=att_name or f"image.{ext}", mime_type=content_type)
if not cached:
logger.warning(
"[teams] Bot Framework attachment '%s' returned data that failed image validation, skipping",
att_name or content_url)
return None
return cached.path, cached.media_type, "image"
path = await cache_image_from_url(content_url)
return (path, content_type, "image") if path else None
except Exception as e:
logger.warning("[teams] Failed to cache image attachment: %s", e)
return None
if content_url: # direct-URL non-image attachment (video/audio/document)
try:
data = await self._fetch_attachment_bytes(content_url)
cached = cache_media_bytes(data, filename=att_name, mime_type=content_type)
return (cached.path, cached.media_type, cached.kind) if cached else None
except Exception as e:
logger.warning("[teams] Failed to cache attachment '%s' (%s): %s", att_name or content_url, content_type, e)
return None
async def _send_card(self, chat_id: str, card: "AdaptiveCard") -> "Any":
"""Send an AdaptiveCard, using a stored ConversationReference when available."""
from microsoft_teams.api import MessageActivityInput
if not self._app:
return None
return await self._send_via_conv_ref(chat_id, MessageActivityInput().add_card(card), card)
@@ -671,24 +584,20 @@ class TeamsAdapter(BasePlatformAdapter):
"""Handle an Adaptive Card Action.Execute button click."""
from tools.approval import resolve_gateway_approval, has_blocking_approval
action = ctx.activity.value.action
data = action.data or {}
data = ctx.activity.value.action.data or {}
hermes_action = data.get("hermes_action", "")
session_key = data.get("session_key", "")
if not hermes_action or not session_key:
return self._invoke_message("Unknown action.")
# Default-deny: approval clicks require TEAMS_ALLOWED_USERS or an explicit
# TEAMS_ALLOW_ALL_USERS=true opt-in, else anyone who can message the bot
# could approve dangerous commands.
# TEAMS_ALLOW_ALL_USERS=true opt-in, else anyone who can message the bot could approve.
allowed_csv = os.getenv("TEAMS_ALLOWED_USERS", "").strip()
allow_all = os.getenv("TEAMS_ALLOW_ALL_USERS", "").strip().lower() in {"1", "true", "yes"}
if not allow_all:
if not allowed_csv:
logger.warning(
"[teams] card action rejected: TEAMS_ALLOWED_USERS not configured "
"and TEAMS_ALLOW_ALL_USERS not set — default deny"
)
"and TEAMS_ALLOW_ALL_USERS not set — default deny")
return self._invoke_message("⛔ Approval buttons require TEAMS_ALLOWED_USERS to be configured.")
from_account = ctx.activity.from_
clicker_id = getattr(from_account, "aad_object_id", None) or getattr(from_account, "id", "")
@@ -702,36 +611,20 @@ class TeamsAdapter(BasePlatformAdapter):
return self._invoke_message("Unknown action.")
if not has_blocking_approval(session_key):
return self._invoke_card([TextBlock(text="⚠️ Approval already resolved or expired.", wrap=True)])
resolve_gateway_approval(session_key, choice)
cmd = data.get("cmd", "")
desc = data.get("desc", "")
body = []
if cmd:
body.append(TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"))
body.append(TextBlock(text=f"```\n{cmd}\n```", wrap=True))
if desc:
body.append(TextBlock(text=f"Reason: {desc}", wrap=True, isSubtle=True))
body = _approval_body(data.get("cmd", ""), data.get("desc", ""))
body.append(TextBlock(text=_APPROVAL_LABELS[choice], wrap=True, weight="Bolder"))
return self._invoke_card(body)
async def send_exec_approval(
self, chat_id: str, command: str, session_key: str, description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None, allow_permanent: bool = True, allow_session: bool = True,
smart_denied: bool = False,
) -> SendResult:
smart_denied: bool = False) -> SendResult:
"""Send an Adaptive Card approval prompt with Allow/Deny buttons."""
if not self._app:
return SendResult(success=False, error="Teams app not initialized")
cmd_preview = command[:2000] + "..." if len(command) > 2000 else command
# Button data carries a truncated cmd — just enough to reconstruct the card body.
btn_data_base = {
"session_key": session_key,
"cmd": command[:200] + "..." if len(command) > 200 else command,
"desc": description,
}
btn_data_base = {"session_key": session_key, "cmd": _truncate(command, 200), "desc": description}
def _action(title: str, hermes_action: str, **kw) -> "ExecuteAction":
return ExecuteAction(
@@ -744,19 +637,13 @@ class TeamsAdapter(BasePlatformAdapter):
if allow_permanent:
actions.append(_action("Always Allow", "approve_always"))
actions.append(_action("Deny", "deny", style="destructive"))
body = [
TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"),
TextBlock(text=f"```\n{cmd_preview}\n```", wrap=True),
TextBlock(text=f"Reason: {description}", wrap=True, isSubtle=True),
]
body = _approval_body(_truncate(command, 2000), description, always=True)
if smart_denied:
body.append(TextBlock(text="Smart DENY: owner override applies to this one operation only.", wrap=True))
card = AdaptiveCard().with_version("1.4").with_body(body).with_actions(actions)
try:
result = await self._send_card(chat_id, card)
message_id = getattr(result, "id", None) if result else None
return SendResult(success=True, message_id=message_id)
return SendResult(success=True, message_id=getattr(result, "id", None) if result else None)
except Exception as e:
logger.error("[teams] send_exec_approval failed: %s", e, exc_info=True)
return SendResult(success=False, error=str(e), retryable=True)
@@ -766,7 +653,6 @@ class TeamsAdapter(BasePlatformAdapter):
) -> SendResult:
if not self._app:
return SendResult(success=False, error="Teams app not initialized")
last_message_id = None
for chunk in self.truncate_message(self.format_message(content)):
try:
@@ -774,8 +660,7 @@ class TeamsAdapter(BasePlatformAdapter):
try:
result = await self._app.reply(chat_id, reply_to, chunk)
except Exception as reply_err:
# Group chats 400 on threaded sends; the SDK has no typed
# HTTP errors, so fall back on any exception.
# Group chats 400 on threaded sends; the SDK has no typed HTTP errors → fall back on any.
logger.debug("Teams reply() failed, falling back to flat send: %s", reply_err)
result = await self._app.send(chat_id, chunk)
else:
@@ -797,11 +682,8 @@ class TeamsAdapter(BasePlatformAdapter):
self, chat_id: str, source: str, default_mime: str, caption: Optional[str] = None, media_label: str = "media"
) -> SendResult:
"""Send any media file/URL as a Teams attachment (shared by send_image/video/voice/document).
Remote ``http(s)://`` URLs are attached by reference; local paths
(optional ``file://`` prefix) are base64-encoded into a data URI.
MIME type is guessed from the path, falling back to ``default_mime``.
"""
Remote ``http(s)://`` URLs are attached by reference; local paths (optional ``file://`` prefix)
are base64-encoded into a data URI. MIME is guessed from the path, else ``default_mime``."""
if not self._app:
return SendResult(success=False, error="Teams app not initialized")
try:
@@ -809,7 +691,7 @@ class TeamsAdapter(BasePlatformAdapter):
import mimetypes
from microsoft_teams.api import Attachment, MessageActivityInput
if source.startswith("http://") or source.startswith("https://"):
if source.startswith(("http://", "https://")):
content_url = source
mime_type = mimetypes.guess_type(source.split("?")[0])[0] or default_mime
else:
@@ -829,8 +711,7 @@ class TeamsAdapter(BasePlatformAdapter):
async def send_image(
self, chat_id: str, image_url: str, caption: Optional[str] = None, reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
metadata: Optional[Dict[str, Any]] = None) -> SendResult:
return await self._send_media_attachment(chat_id, image_url, "image/png", caption=caption, media_label="image")
async def send_image_file(
@@ -840,14 +721,12 @@ class TeamsAdapter(BasePlatformAdapter):
async def send_video(
self, chat_id: str, video_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None, **kwargs,
) -> SendResult:
metadata: Optional[Dict[str, Any]] = None, **kwargs) -> SendResult:
return await self._send_media_attachment(chat_id, video_path, "video/mp4", caption=caption, media_label="video")
async def send_voice(
self, chat_id: str, audio_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None, **kwargs,
) -> SendResult:
metadata: Optional[Dict[str, Any]] = None, **kwargs) -> SendResult:
return await self._send_media_attachment(chat_id, audio_path, "audio/mpeg", caption=caption, media_label="voice")
async def send_document(
@@ -855,20 +734,27 @@ class TeamsAdapter(BasePlatformAdapter):
reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs,
) -> SendResult:
return await self._send_media_attachment(
chat_id, file_path, "application/octet-stream", caption=caption, media_label="document"
)
chat_id, file_path, "application/octet-stream", caption=caption, media_label="document")
async def get_chat_info(self, chat_id: str) -> dict:
return {"name": chat_id, "type": "unknown", "chat_id": chat_id}
# ── Interactive setup ─────────────────────────────────────────────────────────
_SETUP_CREDENTIALS = (
("Client ID", "TEAMS_CLIENT_ID", {}),
("Client secret", "TEAMS_CLIENT_SECRET", {"password": True}),
("Tenant ID", "TEAMS_TENANT_ID", {}),
)
("Tenant ID", "TEAMS_TENANT_ID", {}))
_SETUP_INTRO = (
"You'll need the Teams CLI. If you haven't already:",
" npm install -g @microsoft/teams.cli@preview",
" teams login",
"",
"Then expose port 3978 publicly (devtunnel / ngrok / cloudflared),",
"and create your bot:",
' teams app create --name "Hermes" --endpoint "https://<tunnel>/api/messages"',
"",
"The CLI will print CLIENT_ID, CLIENT_SECRET, and TENANT_ID. Paste them below.",
"")
def interactive_setup() -> None:
@@ -882,17 +768,8 @@ def interactive_setup() -> None:
if not prompt_yes_no("Reconfigure Teams?", False):
return
print_info("You'll need the Teams CLI. If you haven't already:")
print_info(" npm install -g @microsoft/teams.cli@preview")
print_info(" teams login")
print()
print_info("Then expose port 3978 publicly (devtunnel / ngrok / cloudflared),")
print_info("and create your bot:")
print_info(" teams app create --name \"Hermes\" --endpoint \"https://<tunnel>/api/messages\"")
print()
print_info("The CLI will print CLIENT_ID, CLIENT_SECRET, and TENANT_ID. Paste them below.")
print()
for line in _SETUP_INTRO:
print_info(line) if line else print()
for label, env_key, prompt_kwargs in _SETUP_CREDENTIALS:
value = prompt(label, default=get_env_value(env_key) or "", **prompt_kwargs)
if not value:
@@ -919,13 +796,9 @@ def interactive_setup() -> None:
print_info("Restart the gateway: hermes gateway restart")
# ── Plugin entry point ────────────────────────────────────────────────────────
def _install_hint() -> str:
"""Install hint derived from the LAZY_DEPS pins (aiohttp is CVE-pinned, so bumps happen).
``venv_pip=True`` targets the real Hermes venv, sidestepping PEP 668 on Ubuntu 24.04.
"""
"""Install hint derived from the LAZY_DEPS pins (aiohttp is CVE-pinned, so bumps happen);
``venv_pip=True`` targets the real Hermes venv, sidestepping PEP 668 on Ubuntu 24.04."""
try:
from tools.lazy_deps import feature_install_command
cmd = feature_install_command("platform.teams", venv_pip=True)
@@ -961,6 +834,4 @@ def register(ctx) -> None:
"You are chatting via Microsoft Teams. Teams renders a subset of "
"markdown — bold (**text**), italic (*text*), and inline code "
"(`code`) work, but complex tables or raw HTML do not. Keep "
"responses clear and professional."
),
)
"responses clear and professional."))