RCS Providers

Base Classes

RCSProvider

Bases: ABC

RCS delivery provider for rich communication services.

name property

name

Provider name (e.g. 'twilio', 'sinch').

sender_id abstractmethod property

sender_id

RCS sender/agent identifier.

supports_fallback property

supports_fallback

Whether this provider supports automatic SMS fallback.

send abstractmethod async

send(event, to, *, fallback=True)

Send an RCS message.

Parameters:

Name	Type	Description	Default
event	RoomEvent	The room event containing the message content.	required
to	str	Recipient phone number (E.164 format).	required
fallback	bool	If True, allow fallback to SMS when RCS unavailable.	True

Returns:

Type	Description
RCSDeliveryResult	Result with delivery info including whether fallback occurred.

check_capability async

check_capability(phone_number)

Check if a phone number supports RCS.

Parameters:

Name	Type	Description	Default
phone_number	str	Phone number to check (E.164 format).	required

Returns:

Type	Description
bool	True if the number supports RCS, False otherwise.

Raises:

Type	Description
NotImplementedError	If the provider doesn't support capability check.

parse_webhook async

parse_webhook(payload)

Parse an inbound webhook payload into an InboundMessage.

verify_signature

verify_signature(payload, signature, timestamp=None)

Verify that a webhook payload was signed by the provider.

Parameters:

Name	Type	Description	Default
payload	bytes	Raw request body bytes.	required
signature	str	Signature header value from the webhook request.	required
timestamp	str | None	Timestamp header value (required by some providers).	None

Returns:

Type	Description
bool	True if the signature is valid, False otherwise.

close async

close()

Release resources. Override in subclasses that hold connections.

RCSDeliveryResult

Bases: ProviderResult

Extended result for RCS delivery with fallback information.

MockRCSProvider

MockRCSProvider(sender_id='mock_rcs_agent', *, simulate_fallback=False, simulate_failure=False)

Bases: RCSProvider

Mock RCS provider for testing.

Initialize mock provider.

Parameters:

Name	Type	Description	Default
sender_id	str	Mock sender/agent ID.	'mock_rcs_agent'
simulate_fallback	bool	If True, simulate SMS fallback on sends.	False
simulate_failure	bool	If True, simulate send failures.	False

check_capability async

check_capability(phone_number)

Mock capability check - returns opposite of simulate_fallback.

Telnyx

TelnyxRCSProvider

TelnyxRCSProvider(config, public_key=None)

Bases: RCSProvider

RCS provider using the Telnyx REST API.

Telnyx RCS uses the same /v2/messages endpoint as SMS, but requires an RCS agent_id as the sender. When the recipient doesn't support RCS, messages can fall back to SMS (if fallback=True).

Example

config = TelnyxRCSConfig( api_key="KEY...", agent_id="your-rcs-agent-id", ) provider = TelnyxRCSProvider(config) result = await provider.send(event, to="+14155551234")

Initialize the Telnyx RCS provider.

Parameters:

Name	Type	Description	Default
config	TelnyxRCSConfig	Telnyx RCS configuration.	required
public_key	str | None	Telnyx public key for webhook signature verification. Found in Mission Control Portal > Keys & Credentials > Public Key.	None

sender_id property

sender_id

RCS agent ID used as sender.

send async

send(event, to, *, fallback=True)

Send an RCS message via Telnyx.

Parameters:

Name	Type	Description	Default
event	RoomEvent	The room event containing the message content.	required
to	str	Recipient phone number (E.164 format).	required
fallback	bool	If True, allow SMS fallback. If False, RCS only.	True

Returns:

Type	Description
RCSDeliveryResult	Result with delivery info including channel used.

check_capability async

check_capability(phone_number)

Check if a phone number supports RCS.

Parameters:

Name	Type	Description	Default
phone_number	str	Phone number to check (E.164 format).	required

Returns:

Type	Description
bool	True if the number supports RCS, False otherwise.

verify_signature

verify_signature(payload, signature, timestamp=None, *, tolerance_seconds=300)

Verify a Telnyx webhook signature using Ed25519.

Checks both the cryptographic signature and that timestamp is within tolerance_seconds of the current clock — without the freshness check, a captured signed request can be replayed forever.

Parameters:

Name	Type	Description	Default
payload	bytes	Raw request body bytes.	required
signature	str	Value of the Telnyx-Signature-Ed25519 header.	required
timestamp	str | None	Value of the Telnyx-Timestamp header.	None
tolerance_seconds	int	Maximum allowed clock skew between timestamp and the current wall clock. Default 300s.	300

Returns:

Type	Description
bool	True iff signature verifies AND timestamp is fresh.

Raises:

Type	Description
ValueError	If public_key was not provided to the constructor.
ImportError	If PyNaCl is not installed.

close async

close()

Close the HTTP client.

TelnyxRCSConfig

Bases: BaseModel

Telnyx RCS provider configuration.

Attributes:

Name	Type	Description
api_key	SecretStr	Telnyx API key (v2 key starting with KEY...).
agent_id	str	RCS agent ID (obtained after agent onboarding/brand approval).
messaging_profile_id	str | None	Optional messaging profile ID for webhooks.
timeout	float	HTTP request timeout in seconds.

parse_telnyx_rcs_webhook

parse_telnyx_rcs_webhook(payload, channel_id, *, strict=True)

Convert a Telnyx RCS webhook POST body into an InboundMessage.

Telnyx RCS webhooks use JSON format (unlike SMS which can be form-encoded). The webhook structure follows the same pattern as SMS but includes RCS-specific fields like agent_id.

Parameters:

Name	Type	Description	Default
payload	dict[str, Any]	The webhook POST body as a dictionary.	required
channel_id	str	The channel ID to associate with the message.	required
strict	bool	If True (default), raises ValueError for non-inbound webhooks. Set to False to skip validation (not recommended).	True

Returns:

Type	Description
InboundMessage	An InboundMessage ready for process_inbound().

Raises:

Type	Description
ValueError	If strict=True and the webhook is not an inbound message.

Twilio

TwilioRCSProvider

TwilioRCSProvider(config)

Bases: RCSProvider

RCS provider using the Twilio REST API.

Twilio RCS uses the same Messages API as SMS, but requires an RCS-enabled Messaging Service. When the recipient doesn't support RCS, Twilio can automatically fall back to SMS (unless disabled via fallback=False).

send async

send(event, to, *, fallback=True)

Send an RCS message via Twilio.

Parameters:

Name	Type	Description	Default
event	RoomEvent	The room event containing the message content.	required
to	str	Recipient phone number (E.164 format).	required
fallback	bool	If True, allow SMS fallback. If False, RCS only.	True

Returns:

Type	Description
RCSDeliveryResult	Result with delivery info including channel used.

verify_signature

verify_signature(payload, signature, timestamp=None, url=None)

Verify a Twilio webhook signature using HMAC-SHA1.

RCS webhooks arrive from the same Messages API as SMS and are signed identically, so this is the same check the SMS provider runs. It was missing here, which meant an RCS webhook endpoint had no way to establish that Twilio sent it — the inherited method raised NotImplementedError.

Parameters:

Name	Type	Description	Default
payload	bytes	Raw request body bytes (form-encoded).	required
signature	str	Value of the X-Twilio-Signature header.	required
timestamp	str | None	Not used by Twilio (interface compatibility).	None
url	str | None	The exact public URL Twilio called. Required — it is part of the signed string.	None

Returns:

Type	Description
bool	True if the signature is valid, False otherwise.

TwilioRCSConfig

Bases: BaseModel

Twilio RCS provider configuration.

parse_twilio_rcs_webhook

parse_twilio_rcs_webhook(payload, channel_id)

Convert a Twilio RCS webhook POST body into an InboundMessage.

Twilio sends webhooks as form-encoded data. Convert to dict first:

payload = dict(await request.form())