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.

Installation

pip install async-social-bridge

Quick Start

Parse incoming webhooks

from chatbridge.meta import parse_webhook

# In your FastAPI / Flask webhook handler:
payload = await request.json()
messages = parse_webhook(payload)

for msg in messages:
    print(f"[{msg.platform}] {msg.sender_name or msg.sender_id}: {msg.text}")

Send messages

from chatbridge.meta import WhatsAppAdapter, MessengerAdapter

# WhatsApp
async with WhatsAppAdapter(access_token="your_token") as wa:
    result = await wa.send_text(
        receiver_id="your_phone_number_id",
        to="recipient_phone",
        text="Hello from chatbridge!"
    )
    print(result.success)       # True
    print(result.message_id)    # "wamid.xxx..."

# Messenger
async with MessengerAdapter(access_token="your_token") as fb:
    result = await fb.send_text(
        receiver_id="your_page_id",
        to="recipient_psid",
        text="Hello from chatbridge!"
    )

Features

• Async-first — Built on httpx for non-blocking I/O
• Typed models — IncomingMessage and SendResult are Pydantic models with full type hints
• Webhook parsing — One function to normalize WhatsApp, Messenger, and Instagram payloads
• Context manager — Proper HTTP client lifecycle with async with
• Minimal dependencies — Only httpx and pydantic
• Extensible — Subclass BaseAdapter to add Telegram, Slack, or any other platform

Models

IncomingMessage

SendResult

Extending

Create your own adapter by subclassing BaseAdapter:

from chatbridge.base import BaseAdapter
from chatbridge.models import SendResult

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(self, receiver_id: str, to: str, text: str) -> SendResult:
        resp = await self._client.post(
            f"/bot{self.access_token}/sendMessage",
            json={"chat_id": to, "text": text},
        )
        data = resp.json()
        return SendResult(
            success=data.get("ok", False),
            message_id=str(data.get("result", {}).get("message_id")),
            status_code=resp.status_code,
            raw_response=data,
        )

License

MIT