RCS Business Messaging

import base64, hashlib, hmac, json, os
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from fastapi.responses import PlainTextResponse

CLIENT_TOKEN = os.environ["RBM_CLIENT_TOKEN"].encode()   # placeholder: set in the secret store
app = FastAPI()
seen: set[str] = set()                                    # replace with Redis SET NX and a TTL


def valid(header: str, data_b64: str) -> bool:
    digest = hmac.new(CLIENT_TOKEN, base64.b64decode(data_b64), hashlib.sha512).digest()
    return hmac.compare_digest(base64.b64encode(digest).decode(), header)


@app.post("/rbm/webhook")
async def webhook(request: Request, background: BackgroundTasks):
    body = await request.json()
    if "secret" in body and body.get("clientToken") == CLIENT_TOKEN.decode():
        return PlainTextResponse(body["secret"])           # webhook verification handshake
    data = body["message"]["data"]
    if not valid(request.headers.get("X-Goog-Signature", ""), data):
        raise HTTPException(status_code=401)
    event = json.loads(base64.b64decode(data))
    mid = event.get("messageId")
    if mid and mid not in seen:                            # redeliveries are possible
        seen.add(mid)
        background.add_task(answer, event)
    return PlainTextResponse("ok")                         # acknowledge quickly

import asyncio, httpx, uuid
from google.oauth2 import service_account
from google.auth.transport.requests import Request as AuthRequest

SCOPES = ["https://www.googleapis.com/auth/rcsbusinessmessaging"]   # confirm in the REST reference
creds = service_account.Credentials.from_service_account_file(
    os.environ["RBM_SERVICE_ACCOUNT_FILE"], scopes=SCOPES)
AGENT_ID = os.environ["RBM_AGENT_ID"]
BASE = "https://rcsbusinessmessaging.googleapis.com/v1"             # regional hosts also exist
http = httpx.AsyncClient(timeout=10)


async def send(phone: str, content: dict) -> None:
    if not creds.valid:
        await asyncio.to_thread(creds.refresh, AuthRequest())        # blocking call, keep it off the event loop
    response = await http.post(
        f"{BASE}/phones/{phone}/agentMessages",
        params={"messageId": uuid.uuid4().hex, "agentId": AGENT_ID},
        headers={"Authorization": f"Bearer {creds.token}"},
        json={"contentMessage": content, "messageTrafficType": "SERVICEREQUEST"},
    )
    response.raise_for_status()


async def answer(event: dict) -> None:
    phone = event["senderPhoneNumber"]
    text = event.get("text") or event.get("suggestionResponse", {}).get("postbackData", "")
    result = await core_ask(conversation_id=f"rcs:{AGENT_ID}:{phone}", user_id=pseudonym(phone), text=text)
    for content in to_rbm_content(result):   # whole messages, in order: RCS has no token streaming
        await send(phone, content)

@RestController
class RbmWebhook {
  private final byte[] clientToken = System.getenv("RBM_CLIENT_TOKEN").getBytes(StandardCharsets.UTF_8);
  private final ObjectMapper mapper = new ObjectMapper();
  private final RbmWorker worker;                       // queue-backed, wraps the Spring AI core

  RbmWebhook(RbmWorker worker) { this.worker = worker; }

  @PostMapping("/rbm/webhook")
  ResponseEntity<String> receive(@RequestBody JsonNode body,
      @RequestHeader(value = "X-Goog-Signature", required = false) String signature) throws Exception {
    if (body.has("secret") && body.path("clientToken").asText().equals(new String(clientToken, UTF_8))) {
      return ResponseEntity.ok(body.get("secret").asText());       // handshake
    }
    byte[] payload = Base64.getDecoder().decode(body.path("message").path("data").asText());
    Mac mac = Mac.getInstance("HmacSHA512");
    mac.init(new SecretKeySpec(clientToken, "HmacSHA512"));
    byte[] expected = Base64.getEncoder().encode(mac.doFinal(payload));
    if (signature == null || !MessageDigest.isEqual(expected, signature.getBytes(UTF_8))) {
      return ResponseEntity.status(401).build();
    }
    worker.submit(mapper.readTree(payload));            // returns immediately; ChatClient runs on the worker
    return ResponseEntity.ok("ok");
  }
}