Receive SMS

Python

"""Receive inbound SMS and delivery receipts by webhook (pip install fastapi uvicorn).

Run with:  uvicorn webhooks:app --host 0.0.0.0 --port 8000
"""
import os
import secrets
from datetime import datetime

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()

# A long random token in the URL stops anyone else posting to your endpoints.
TOKEN = os.environ["EM_WEBHOOK_TOKEN"]


class InboundMessage(BaseModel):
    type: str
    messageId: int
    sender: str
    inboxName: str | None = None
    inboxNumber: str | None = None
    inboxKeyword: str | None = None
    content: str
    dateTime: datetime


class DeliveryReceipt(BaseModel):
    type: str
    messageId: int
    reference: str | None = None
    recipient: str
    status: str
    dateTime: datetime


class InboundBatch(BaseModel):
    items: list[InboundMessage] = Field(alias="list")


class ReceiptBatch(BaseModel):
    items: list[DeliveryReceipt] = Field(alias="list")


def check(key: str) -> None:
    if not secrets.compare_digest(key, TOKEN):
        raise HTTPException(status_code=404)


@app.post("/everymessage/inbound/{key}")
def inbound(key: str, batch: InboundBatch):
    check(key)
    for m in batch.items:
        print(f"SMS {m.messageId} from {m.sender} to {m.inboxNumber}: {m.content}")
    # Any 2xx response tells everymessage the batch was received.
    return {"echo": "ACK"}


@app.post("/everymessage/receipts/{key}")
def receipts(key: str, batch: ReceiptBatch):
    check(key)
    for r in batch.items:
        print(f"Message {r.messageId} ({r.reference}) to {r.recipient}: {r.status}")
    return {"echo": "ACK"}

Python

"""Verify everymessage webhook signatures (X-EM-Signature-V2) with FastAPI.

pip install fastapi uvicorn, then run:  uvicorn verify_signature:app --host 0.0.0.0 --port 8000
"""
import hashlib
import hmac
import json
import os
import time

from fastapi import FastAPI, HTTPException, Request

app = FastAPI()

# The signing secret shown once when the webhook was created (Admin > Webhooks).
SECRET = os.environ["EM_WEBHOOK_SECRET"].encode()


def verify(raw_body: bytes, timestamp: str, signature_header: str) -> bool:
    # Reject anything sent more than 5 minutes ago (or in the future) to stop replays.
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > 300:
        return False
    # HMAC-SHA256 of "{timestamp}.{body}", lowercase hex.
    expected = hmac.new(SECRET, timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature_header.removeprefix("v2="), expected)


@app.post("/everymessage/webhook")
async def webhook(request: Request):
    # Verify the exact bytes received: parsing and re-serialising the JSON would change them.
    raw_body = await request.body()
    if not verify(raw_body,
                  request.headers.get("X-EM-Timestamp", ""),
                  request.headers.get("X-EM-Signature-V2", "")):
        raise HTTPException(status_code=401)

    # Signature checked: now parse and process the JSON.
    print("Verified webhook:", json.loads(raw_body))
    return {"ok": True}

Node.js

// Verify everymessage webhook signatures (X-EM-Signature-V2) with Express.
// npm install express, then run:  node verify_signature.js
const crypto = require('crypto');
const express = require('express');

// The signing secret shown once when the webhook was created (Admin > Webhooks).
const SECRET = process.env.EM_WEBHOOK_SECRET;

function verify(rawBody, timestamp, signatureHeader) {
  // Reject anything sent more than 5 minutes ago (or in the future) to stop replays.
  if (!/^d+$/.test(timestamp || '') || Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false;
  // HMAC-SHA256 of "{timestamp}.{body}", lowercase hex.
  const expected = crypto.createHmac('sha256', SECRET).update(timestamp + '.').update(rawBody).digest('hex');
  const received = (signatureHeader || '').replace(/^v2=/, '');
  return received.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(received), Buffer.from(expected));
}

const app = express();

// express.raw keeps the exact bytes received: parsing and re-serialising the JSON would change them.
app.post('/everymessage/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  if (!verify(req.body, req.get('X-EM-Timestamp'), req.get('X-EM-Signature-V2'))) {
    return res.sendStatus(401);
  }
  // Signature checked: now parse and process the JSON.
  console.log('Verified webhook:', JSON.parse(req.body));
  res.sendStatus(200);
});

app.listen(8000);

Python

"""Collect inbound SMS replies with the everymessage v1 API (Python 3.10+, pip install httpx)."""
import os
import time

import httpx

with httpx.Client(
    base_url="https://api.everymessage.com",
    auth=(os.environ["EM_API_KEY"], os.environ["EM_SECRET_KEY"]),
    timeout=30,
) as client:
    while True:
        # Each message is returned once: store it before the next call.
        response = client.get("/v1/inbound_messages", params={"limit": 500})
        response.raise_for_status()
        result = response.json()

        reason = result.get("responseStatusReason")
        if reason is None:
            for m in result["body"]:
                print(f"{m['dateTime']} from {m['sender']} to {m['inboxNumber']}: {m['content']}")
        elif reason != "NO_CURRENT_ENTRIES":
            raise RuntimeError(f"Collect failed: {reason}")

        time.sleep(60)