async-social-bridge

Async messaging adapters for Meta platforms (WhatsApp, Messenger, Instagram).
Send and parse messages with a clean, typed Python API. Zero bloat — only httpx and pydantic as dependencies.

Features

• Async First — Built on httpx for high-performance, non-blocking I/O
• Fully Typed — Pydantic models with full IDE autocomplete and runtime validation
• Unified API — One adapter and one parser for all Meta platforms. No platform-specific spaghetti code
• Zero Boilerplate — Automatically extracts platform, sender_id, and text from messy webhook payloads
• Multi-Tenant Ready — Dynamic access tokens per request; ideal for SaaS platforms managing multiple client accounts
• Extensible — Subclass BaseAdapter and BaseParser to add any platform (Telegram, Slack, Discord, etc.)

Installation

pip install async-social-bridge

Quick Start: A Complete Example

Here is how async-social-bridge fits into a real FastAPI webhook endpoint — from raw payload to AI reply.

from fastapi import FastAPI, Request
from socialbridge.meta import MetaParser, MetaAdapter
from socialbridge.models import OutgoingMessage

app = FastAPI()
parser = MetaParser()

@app.post("/webhook")
async def webhook(request: Request):
    # Step 1: Get the raw JSON from Meta
    payload = await request.json()

    # Step 2: Parse the messy nested payload into clean Python objects
    messages = parser.parse(payload)

    for msg in messages:
        # msg.platform     -> "whatsapp", "messenger", or "instagram"
        # msg.sender_id    -> the user who messaged you
        # msg.receiver_id  -> your WhatsApp Phone ID or Facebook Page ID
        # msg.text         -> the plain text content

        # Step 3: Look up your tenant's token using the receiver_id
        # access_token = await db.get_token(msg.receiver_id)
        access_token = "YOUR_ACCESS_TOKEN"

        # Step 4: Build the reply
        reply = OutgoingMessage(
            platform=msg.platform,
            receiver_id=msg.receiver_id,
            to=msg.sender_id,
            text=f"You said: {msg.text}"
        )

        # Step 5: Send the reply — one API for all platforms
        async with MetaAdapter(access_token=access_token) as meta:
            result = await meta.send_text_message(reply)
            if result.success:
                print(f"Sent! ID: {result.metadata['message_id']}")

    return {"status": "ok"}

Data Models Reference

IncomingMessage

Returned by MetaParser.parse(). You never construct this yourself — the library creates it from the raw webhook payload.

OutgoingMessage

Passed to MetaAdapter.send_text_message(). You create this yourself with your reply data.

SendResult

Returned by MetaAdapter.send_text_message().

Error Handling

The library raises specific exceptions so you can handle failures precisely without catching generic Exception:

from socialbridge.exceptions import ParseError, SendError

# ParseError: raised when the webhook payload is not a recognized Meta format
try:
    messages = parser.parse(payload)
except ParseError as e:
    print(f"Unrecognized payload: {e}")
    return {"status": "ignored"}

# SendError: raised when Meta's API returns a non-200 response
try:
    result = await meta.send_text_message(reply)
except SendError as e:
    print(f"Failed to send. HTTP {e.status_code}: {e}")

Extending: Adding a New Platform

Create a new platform adapter by subclassing BaseAdapter and BaseParser:

from socialbridge.base import BaseAdapter, BaseParser
from socialbridge.models import IncomingMessage, OutgoingMessage, SendResult
from socialbridge.exceptions import SendError


class TelegramParser(BaseParser):
    def parse(self, payload: dict) -> list[IncomingMessage]:
        messages = []
        for update in payload.get("updates", []):
            msg = update.get("message", {})
            messages.append(IncomingMessage(
                platform="telegram",
                sender_id=str(msg["from"]["id"]),
                receiver_id=str(msg["chat"]["id"]),
                text=msg.get("text", ""),
                raw=update,
            ))
        return messages


class TelegramAdapter(BaseAdapter):
    def __init__(self, bot_token: str):
        super().__init__(
            access_token=bot_token,
            base_url="https://api.telegram.org",
        )

    async def send_text_message(self, message: OutgoingMessage) -> SendResult:
        resp = await self._client.post(
            f"/bot{self.access_token}/sendMessage",
            json={"chat_id": message.to, "text": message.text},
        )
        if resp.status_code != 200:
            raise SendError(
                message=f"Telegram API error",
                status_code=resp.status_code,
                response_body=resp.text,
            )
        data = resp.json()
        return SendResult(
            success=True,
            metadata={"message_id": str(data["result"]["message_id"])},
            status_code=resp.status_code,
            raw_response=data,
        )

Supported Platforms

• WhatsApp Business Cloud
• Facebook Messenger
• Instagram Direct

License

MIT