Receiving RCS Messages with the Python SDK

Prerequisites

Before proceeding, ensure you have obtained an RCS sandbox agent and API key as described in the prerequisites.

Installation

Create a Python virtual environment:

Shell

python3 -m venv venv

Activate the virtual environment:

Shell

source venv/bin/activate

Install the Pinnacle Python SDK, FastAPI, and python-dotenv:

Shell

pip install rcs==2.0.28 "fastapi[standard]" python-dotenv

This guide uses version rcs==2.0.28. Use Python 3.10 or later for this FastAPI example (the SDK itself supports Python 3.8+)

Configuration

Create an .env file in your project root and add your Pinnacle API key and signing secret:

Shell

PINNACLE_API_KEY="your_api_key" # pnclk_
AGENT_ID="your_agent_id" # agent_
PINNACLE_SIGNING_SECRET="your_signing_secret" # pss_

Setting Up a Webhook

To receive inbound RCS messages, you need to configure a webhook in the Pinnacle dashboard:

Navigate to Development > Webhooks in the Pinnacle dashboard
Click Create new webhook
Give your webhook a descriptive name
Enter your webhook endpoint URL
- For local development, use an ngrok tunnel pointing to localhost:8000/inbound-rcs
- For production, use your deployed server URL
After creation, copy the signing secret and add it to your .env file
Add your RCS sandbox agent to your webhook for it to receive messages. You must also whitelist the devices you want to test with by navigating to your sandbox agent and adding test device phone numbers.

Optionally, you can configure custom HTTP headers (e.g. X-API-KEY) to be sent on every webhook delivery. Add them in the dashboard or via the headers field on POST /webhooks/attach. The PINNACLE-SIGNING-SECRET header is reserved.

Creating Your Webhook Endpoint

Create a new Python file (e.g., main.py) and use the complete example.

The complete example creates a FastAPI endpoint that:

Receives webhook POST requests at /inbound-rcs
Verifies the webhook signature using your signing secret
Processes received messages, message status updates, typing status updates, and data attached to a button clicked by the recipient.

Running Your Server

Start the FastAPI server:

Shell

fastapi dev main.py

Your server will start on http://localhost:8000. If you're using ngrok for local development, start it in a separate terminal:

Shell

ngrok http 8000

Use the ngrok URL (e.g., https://abc123.ngrok.io/inbound-rcs) as your webhook endpoint in the Pinnacle dashboard.

Testing Your Webhook

Send a POST request using your whitelisted test number. This unauthenticated demo endpoint is for local testing only.

Shell

curl -X POST "http://localhost:8000/send-rcs/+14155550123"

If there are no errors, you should see something like

JSON

{
  "message_ids": null,
  "segments": 1,
  "total_cost": 0.03,
  "sender": "agent_exampleId",
  "recipient": "+12345678910",
  "status": "queued",
  "messageId": 6828
}

and receive a message on your whitelisted device like this: rcs message

If you're not receiving any messages, make sure you have

Your RCS sandbox agent associated with your webhook
Your test device is whitelisted

If you tap "Say hello back", your whitelisted device should receive a text saying "Hello! Button clicked successfully."

With that, your webhook should now be successfully receiving inbound RCS messages as well message status updates for outbound messages! You can monitor these statuses by filtering events by their message direction set to outbound:

Python

if message_event.direction == "OUTBOUND":
    print(message_event)

For more detail about processing the message payload received, please view the process method.

Complete example

Python

from fastapi import FastAPI, HTTPException, Request
from rcs import (
    AsyncPinnacle,
    MessageEvent,
    SendRichCardsOptions,
    RcsButtonSendLocationLatLong,
    RichCard,
    RichButton_Call,
    RichButton_OpenUrl,
    RichButton_RequestUserLocation,
    RichButton_ScheduleEvent,
    RichButton_SendLocation,
    RichButton_Trigger,
    RichCardsMessage,
    RichTextMessage,
)
from dotenv import load_dotenv
import os
from datetime import datetime, timedelta, timezone

load_dotenv()

app = FastAPI()

client = AsyncPinnacle(api_key=os.environ.get("PINNACLE_API_KEY"))


@app.post("/send-rcs/{phone_number}")
async def send_rcs(phone_number: str):
    res = await client.messages.rcs.send(
        request=RichCardsMessage(
            options=SendRichCardsOptions(
                validate_=True
            ),  # Will let you know if your RCS message will fail to be sent ahead of making the request to carriers
            from_=os.getenv("AGENT_ID"),
            to=phone_number,
            cards=[
                RichCard(
                    media="https://server.trypinnacle.app/storage/v1/object/sign/vault/3/67330f56-8fc4-4d9e-882d-161b84fc8e31/Your_image_here.png?token=eyJraWQiOiJzdG9yYWdlLXVybC1zaWduaW5nLWtleV9hOGI0YTI0NC00NzY4LTRhOTktYWI4MS1iNmZhNTZhNGQyZWYiLCJhbGciOiJIUzI1NiJ9.eyJ1cmwiOiJ2YXVsdC8zLzY3MzMwZjU2LThmYzQtNGQ5ZS04ODJkLTE2MWI4NGZjOGUzMS9Zb3VyX2ltYWdlX2hlcmUucG5nIiwiaWF0IjoxNzYwOTg0OTEwLCJleHAiOjMxNzEyMDk4NDkxMH0.bcIMtiBAvV8C7Gw7uYaR5TMGsCep7w1TvRMjoRlP_-g",
                    title="Hello, world!",
                    subtitle="This is an example RCS message with rich content.",
                    buttons=[
                        RichButton_Trigger(
                            type="trigger",
                            metadata="",
                            payload="HELLO",
                            title="Say hello back",
                        ),
                        RichButton_SendLocation(
                            type="sendLocation",
                            lat_long=RcsButtonSendLocationLatLong(
                                lat=36.7749, lng=-122.4194
                            ),
                            metadata="",
                            title="View example location",
                        ),
                    ],
                ),
                RichCard(
                    media="https://server.trypinnacle.app/storage/v1/object/sign/vault/3/67330f56-8fc4-4d9e-882d-161b84fc8e31/Your_image_here.png?token=eyJraWQiOiJzdG9yYWdlLXVybC1zaWduaW5nLWtleV9hOGI0YTI0NC00NzY4LTRhOTktYWI4MS1iNmZhNTZhNGQyZWYiLCJhbGciOiJIUzI1NiJ9.eyJ1cmwiOiJ2YXVsdC8zLzY3MzMwZjU2LThmYzQtNGQ5ZS04ODJkLTE2MWI4NGZjOGUzMS9Zb3VyX2ltYWdlX2hlcmUucG5nIiwiaWF0IjoxNzYwOTg0OTEwLCJleHAiOjMxNzEyMDk4NDkxMH0.bcIMtiBAvV8C7Gw7uYaR5TMGsCep7w1TvRMjoRlP_-g",
                    title="Your second card",
                    subtitle="This subtitle is optional. Each card can have different buttons, like this one allowing you to share your location instead of view a location.",
                    buttons=[
                        RichButton_RequestUserLocation(
                            type="requestUserLocation",
                            metadata="",
                            title="Share your location",
                        ),
                    ],
                ),
            ],
            # Quick replies are available across all cards while buttons are specific to certain cards, and only show when a user swipes to that particular card
            quick_replies=[
                RichButton_ScheduleEvent(
                    type="scheduleEvent",
                    title="Add example event to cal",
                    metadata="",
                    event_title="Sample Event",
                    event_start_time=(datetime.now(timezone.utc) + timedelta(days=1)).isoformat(),
                    event_end_time=(datetime.now(timezone.utc) + timedelta(days=1, hours=1)).isoformat(),
                    event_description="This is an example calendar event.",
                ),
                RichButton_Call(
                    type="call",
                    metadata="",
                    payload="+14152321234",
                    title="Call example number",
                ),
                RichButton_OpenUrl(
                    type="openUrl",
                    metadata="",
                    payload="https://docs.pinnacle.sh/api-reference/messages/send-rcs",
                    title="View RCS docs",
                ),
            ],
        )
    )
    return res


@app.post("/inbound-rcs")
async def inbound_rcs(request: Request):
    try:
        body = await request.body()
        pinnacle_request = {
            "headers": dict(request.headers),
            "body": body,
        }
        # Process and validate the webhook
        # Returns a fully typed MessageEvent object
        message_event: MessageEvent = await client.messages.process(
            pinnacle_request, secret=os.getenv("PINNACLE_SIGNING_SECRET")
        )
        print(message_event)

        # Check if the message contains a button click with HELLO payload
        # View full docs on processing events here: https://docs.pinnacle.sh/methods/process
        if isinstance(message_event, MessageEvent) and message_event.type == "MESSAGE.RECEIVED" and message_event.direction == "INBOUND":
            # Check if this is RCS button data (discriminated by type field)
            if (
                message_event.message.type == "RCS_BUTTON_DATA"
            ):
                button = message_event.message.button
                if button.payload == "HELLO":
                    await send_hello(message_event.conversation.from_)

    except Exception as e:
        print("Error occurred processing webhook message", e)
        raise HTTPException(status_code=400, detail="Unable to process webhook") from e

    return {"message": "You received a message"}


async def send_hello(to: str):
    res = await client.messages.rcs.send(
        request=RichTextMessage(
            text="Hello! Button clicked successfully.",
            from_=os.getenv("AGENT_ID"),
            to=to,
            quick_replies=[],
        )
    )
    print(res)

Receiving RCS Messages

PrerequisitesCopy Anchor Link

InstallationCopy Anchor Link

ConfigurationCopy Anchor Link

Setting Up a WebhookCopy Anchor Link

Creating Your Webhook EndpointCopy Anchor Link

Running Your ServerCopy Anchor Link

Testing Your WebhookCopy Anchor Link

Prerequisites

Installation

Configuration

Setting Up a Webhook

Creating Your Webhook Endpoint

Running Your Server

Testing Your Webhook