feat: add SMS (Twilio) platform adapter
Add SMS as a first-class messaging platform via the Twilio API. Shares credentials with the existing telephony skill — same TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_PHONE_NUMBER env vars. Adapter (gateway/platforms/sms.py): - aiohttp webhook server for inbound (Twilio form-encoded POSTs) - Twilio REST API with Basic auth for outbound - Markdown stripping, smart chunking at 1600 chars - Echo loop prevention, phone number redaction in logs Integration (13 files): - gateway config, run, channel_directory - agent prompt_builder (SMS platform hint) - cron scheduler, cronjob tools - send_message_tool (_send_sms via Twilio API) - toolsets (hermes-sms + hermes-gateway) - gateway setup wizard, status display - pyproject.toml (sms optional extra) - 21 tests Docs: - website/docs/user-guide/messaging/sms.md (full setup guide) - Updated messaging index (architecture, toolsets, security, links) - Updated environment-variables.md reference Inspired by PR #1575 (@sunsakis), rewritten for Twilio.
This commit is contained in:
parent
3d38d85287
commit
07549c967a
16 changed files with 796 additions and 5 deletions
215
tests/gateway/test_sms.py
Normal file
215
tests/gateway/test_sms.py
Normal file
|
|
@ -0,0 +1,215 @@
|
|||
"""Tests for SMS (Twilio) platform integration.
|
||||
|
||||
Covers config loading, format/truncate, echo prevention,
|
||||
requirements check, and toolset verification.
|
||||
"""
|
||||
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.config import Platform, PlatformConfig, HomeChannel
|
||||
|
||||
|
||||
# ── Config loading ──────────────────────────────────────────────────
|
||||
|
||||
class TestSmsConfigLoading:
|
||||
"""Verify _apply_env_overrides wires SMS correctly."""
|
||||
|
||||
def test_sms_platform_enum_exists(self):
|
||||
assert Platform.SMS.value == "sms"
|
||||
|
||||
def test_env_overrides_create_sms_config(self):
|
||||
from gateway.config import load_gateway_config
|
||||
|
||||
env = {
|
||||
"TWILIO_ACCOUNT_SID": "ACtest123",
|
||||
"TWILIO_AUTH_TOKEN": "token_abc",
|
||||
"TWILIO_PHONE_NUMBER": "+15551234567",
|
||||
}
|
||||
with patch.dict(os.environ, env, clear=False):
|
||||
config = load_gateway_config()
|
||||
assert Platform.SMS in config.platforms
|
||||
pc = config.platforms[Platform.SMS]
|
||||
assert pc.enabled is True
|
||||
assert pc.api_key == "token_abc"
|
||||
|
||||
def test_env_overrides_set_home_channel(self):
|
||||
from gateway.config import load_gateway_config
|
||||
|
||||
env = {
|
||||
"TWILIO_ACCOUNT_SID": "ACtest123",
|
||||
"TWILIO_AUTH_TOKEN": "token_abc",
|
||||
"TWILIO_PHONE_NUMBER": "+15551234567",
|
||||
"SMS_HOME_CHANNEL": "+15559876543",
|
||||
"SMS_HOME_CHANNEL_NAME": "My Phone",
|
||||
}
|
||||
with patch.dict(os.environ, env, clear=False):
|
||||
config = load_gateway_config()
|
||||
hc = config.platforms[Platform.SMS].home_channel
|
||||
assert hc is not None
|
||||
assert hc.chat_id == "+15559876543"
|
||||
assert hc.name == "My Phone"
|
||||
assert hc.platform == Platform.SMS
|
||||
|
||||
def test_sms_in_connected_platforms(self):
|
||||
from gateway.config import load_gateway_config
|
||||
|
||||
env = {
|
||||
"TWILIO_ACCOUNT_SID": "ACtest123",
|
||||
"TWILIO_AUTH_TOKEN": "token_abc",
|
||||
}
|
||||
with patch.dict(os.environ, env, clear=False):
|
||||
config = load_gateway_config()
|
||||
connected = config.get_connected_platforms()
|
||||
assert Platform.SMS in connected
|
||||
|
||||
|
||||
# ── Format / truncate ───────────────────────────────────────────────
|
||||
|
||||
class TestSmsFormatAndTruncate:
|
||||
"""Test SmsAdapter.format_message strips markdown."""
|
||||
|
||||
def _make_adapter(self):
|
||||
from gateway.platforms.sms import SmsAdapter
|
||||
|
||||
env = {
|
||||
"TWILIO_ACCOUNT_SID": "ACtest",
|
||||
"TWILIO_AUTH_TOKEN": "tok",
|
||||
"TWILIO_PHONE_NUMBER": "+15550001111",
|
||||
}
|
||||
with patch.dict(os.environ, env):
|
||||
pc = PlatformConfig(enabled=True, api_key="tok")
|
||||
adapter = object.__new__(SmsAdapter)
|
||||
adapter.config = pc
|
||||
adapter._platform = Platform.SMS
|
||||
adapter._account_sid = "ACtest"
|
||||
adapter._auth_token = "tok"
|
||||
adapter._from_number = "+15550001111"
|
||||
return adapter
|
||||
|
||||
def test_strips_bold(self):
|
||||
adapter = self._make_adapter()
|
||||
assert adapter.format_message("**hello**") == "hello"
|
||||
|
||||
def test_strips_italic(self):
|
||||
adapter = self._make_adapter()
|
||||
assert adapter.format_message("*world*") == "world"
|
||||
|
||||
def test_strips_code_blocks(self):
|
||||
adapter = self._make_adapter()
|
||||
result = adapter.format_message("```python\nprint('hi')\n```")
|
||||
assert "```" not in result
|
||||
assert "print('hi')" in result
|
||||
|
||||
def test_strips_inline_code(self):
|
||||
adapter = self._make_adapter()
|
||||
assert adapter.format_message("`code`") == "code"
|
||||
|
||||
def test_strips_headers(self):
|
||||
adapter = self._make_adapter()
|
||||
assert adapter.format_message("## Title") == "Title"
|
||||
|
||||
def test_strips_links(self):
|
||||
adapter = self._make_adapter()
|
||||
assert adapter.format_message("[click](https://example.com)") == "click"
|
||||
|
||||
def test_collapses_newlines(self):
|
||||
adapter = self._make_adapter()
|
||||
result = adapter.format_message("a\n\n\n\nb")
|
||||
assert result == "a\n\nb"
|
||||
|
||||
|
||||
# ── Echo prevention ────────────────────────────────────────────────
|
||||
|
||||
class TestSmsEchoPrevention:
|
||||
"""Adapter should ignore messages from its own number."""
|
||||
|
||||
def test_own_number_detection(self):
|
||||
"""The adapter stores _from_number for echo prevention."""
|
||||
from gateway.platforms.sms import SmsAdapter
|
||||
|
||||
env = {
|
||||
"TWILIO_ACCOUNT_SID": "ACtest",
|
||||
"TWILIO_AUTH_TOKEN": "tok",
|
||||
"TWILIO_PHONE_NUMBER": "+15550001111",
|
||||
}
|
||||
with patch.dict(os.environ, env):
|
||||
pc = PlatformConfig(enabled=True, api_key="tok")
|
||||
adapter = SmsAdapter(pc)
|
||||
assert adapter._from_number == "+15550001111"
|
||||
|
||||
|
||||
# ── Requirements check ─────────────────────────────────────────────
|
||||
|
||||
class TestSmsRequirements:
|
||||
def test_check_sms_requirements_missing_sid(self):
|
||||
from gateway.platforms.sms import check_sms_requirements
|
||||
|
||||
env = {"TWILIO_AUTH_TOKEN": "tok"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
assert check_sms_requirements() is False
|
||||
|
||||
def test_check_sms_requirements_missing_token(self):
|
||||
from gateway.platforms.sms import check_sms_requirements
|
||||
|
||||
env = {"TWILIO_ACCOUNT_SID": "ACtest"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
assert check_sms_requirements() is False
|
||||
|
||||
def test_check_sms_requirements_both_set(self):
|
||||
from gateway.platforms.sms import check_sms_requirements
|
||||
|
||||
env = {
|
||||
"TWILIO_ACCOUNT_SID": "ACtest",
|
||||
"TWILIO_AUTH_TOKEN": "tok",
|
||||
}
|
||||
with patch.dict(os.environ, env, clear=False):
|
||||
# Only returns True if aiohttp is also importable
|
||||
result = check_sms_requirements()
|
||||
try:
|
||||
import aiohttp # noqa: F401
|
||||
assert result is True
|
||||
except ImportError:
|
||||
assert result is False
|
||||
|
||||
|
||||
# ── Toolset verification ───────────────────────────────────────────
|
||||
|
||||
class TestSmsToolset:
|
||||
def test_hermes_sms_toolset_exists(self):
|
||||
from toolsets import get_toolset
|
||||
|
||||
ts = get_toolset("hermes-sms")
|
||||
assert ts is not None
|
||||
assert "tools" in ts
|
||||
|
||||
def test_hermes_sms_in_gateway_includes(self):
|
||||
from toolsets import get_toolset
|
||||
|
||||
gw = get_toolset("hermes-gateway")
|
||||
assert gw is not None
|
||||
assert "hermes-sms" in gw["includes"]
|
||||
|
||||
def test_sms_platform_hint_exists(self):
|
||||
from agent.prompt_builder import PLATFORM_HINTS
|
||||
|
||||
assert "sms" in PLATFORM_HINTS
|
||||
assert "concise" in PLATFORM_HINTS["sms"].lower()
|
||||
|
||||
def test_sms_in_scheduler_platform_map(self):
|
||||
"""Verify cron scheduler recognizes 'sms' as a valid platform."""
|
||||
# Just check the Platform enum has SMS — the scheduler imports it dynamically
|
||||
assert Platform.SMS.value == "sms"
|
||||
|
||||
def test_sms_in_send_message_platform_map(self):
|
||||
"""Verify send_message_tool recognizes 'sms'."""
|
||||
# The platform_map is built inside _handle_send; verify SMS enum exists
|
||||
assert hasattr(Platform, "SMS")
|
||||
|
||||
def test_sms_in_cronjob_deliver_description(self):
|
||||
"""Verify cronjob_tools mentions sms in deliver description."""
|
||||
from tools.cronjob_tools import CRONJOB_SCHEMA
|
||||
deliver_desc = CRONJOB_SCHEMA["parameters"]["properties"]["deliver"]["description"]
|
||||
assert "sms" in deliver_desc.lower()
|
||||
Loading…
Add table
Add a link
Reference in a new issue