Skip to content

twilio_sms

twilio_sms

Twilio SMS channel for sending and receiving text messages.

Classes

TwilioSMSChannel 

TwilioSMSChannel(*, account_sid: str = '', auth_token: str = '', phone_number: str = '', bus: Optional[EventBus] = None)

Bases: BaseChannel

Send and receive SMS via Twilio.

Sending uses the Twilio REST API. Receiving is handled by the /webhooks/twilio FastAPI endpoint which calls registered handlers.

Source code in src/openjarvis/channels/twilio_sms.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    *,
    account_sid: str = "",
    auth_token: str = "",
    phone_number: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._account_sid = account_sid or os.environ.get("TWILIO_ACCOUNT_SID", "")
    self._auth_token = auth_token or os.environ.get("TWILIO_AUTH_TOKEN", "")
    self._phone_number = phone_number or os.environ.get("TWILIO_PHONE_NUMBER", "")
    self._bus = bus
    self._client: Any = None
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
Functions
handle_webhook 
handle_webhook(from_number: str, body: str, message_sid: str = '') -> None

Called by the webhook route for incoming SMS.

Source code in src/openjarvis/channels/twilio_sms.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def handle_webhook(
    self,
    from_number: str,
    body: str,
    message_sid: str = "",
) -> None:
    """Called by the webhook route for incoming SMS."""
    msg = ChannelMessage(
        channel="twilio",
        sender=from_number,
        content=body,
        message_id=message_sid,
    )
    for handler in self._handlers:
        handler(msg)
    if self._bus:
        self._bus.publish(
            EventType.CHANNEL_MESSAGE_RECEIVED,
            {
                "channel": "twilio",
                "sender": from_number,
                "content": body,
                "message_id": message_sid,
            },
        )