ai-prishtina-whatsapp-mcp 0.1.3

Professional WhatsApp Business API library with AI provider integration and MCP support

ai-prishtina-whatsapp-mcp

The WhatsApp Business API library that does everything — except make coffee. (We're working on that.)

A production-ready Python library for WhatsApp Business API integration, featuring a full MCP (Model Context Protocol) server that exposes WhatsApp as AI tools for Claude, Windsurf, Cursor, and any other MCP-compatible client. Built with async-first design, immutable domain models, 9 AI providers, and enough features to make your competitors cry into their lukewarm coffee.

Think of it as the Swiss Army knife of WhatsApp integrations — but one where every blade is a production-grade, type-safe, async-native tool. Including the corkscrew.

---

What's New

MCP Server — WhatsApp as AI Tools

The centerpiece addition: a FastMCP server that exposes every WhatsApp operation as a tool, usable directly from your AI assistant. Configure it once, and suddenly your Claude/Windsurf/Cursor installation can send WhatsApp messages, analyze sentiment, and check provider health — all without writing a line of integration code.

{
  "mcpServers": {
    "whatsapp": {
      "command": "whatsapp-mcp",
      "env": {
        "WHATSAPP_ACCESS_TOKEN": "your_token",
        "WHATSAPP_PHONE_NUMBER_ID": "your_phone_id"
      }
    }
  }
}

Once connected, your AI assistant gets 16 tools. It's like giving your AI a WhatsApp account. Use responsibly.

MCP Tools Available

Tool	What It Does
send_text_message	Send a text message to any WhatsApp number
send_bulk_messages	Blast to multiple recipients with built-in rate limiting
send_media_message	Images, documents, audio, video
send_template_message	Approved template messages with components
send_interactive_message	Buttons and list messages
send_reaction	React to messages with emoji
mark_as_read	Trigger read receipts programmatically
parse_webhook	Parse incoming webhook payloads
verify_webhook_signature	Validate that webhooks actually came from Meta
analyze_sentiment	AI-powered sentiment analysis
detect_intent	Intent detection from message text
translate_message	Translate to any language
get_conversation_history	Sliding-window context per phone number
add_to_conversation	Add messages to the conversation buffer
health_check_providers	Check AI provider health across all registered providers
upload_media / download_media	Media management

---

Features

• 9 AI Providers: OpenAI, Anthropic, Google Gemini, Azure, AWS Bedrock, Ollama (local), Whisper (local), HuggingFace, spaCy — with automatic failover
• MCP Server: Full FastMCP server exposing everything as AI tools
• Functional Architecture: Immutable frozen dataclasses, Result<T,E> monad, no exceptions in business logic
• Rate Limiter: Async token-bucket rate limiter (Meta will thank you)
• Bulk Messaging: Multi-recipient dispatch with per-result reporting
• Conversation Buffer: Sliding-window context per phone number for LLM-ready history
• Message Pipeline: Composable async middleware chain for webhook event processing
• Message Deduplication: Redis/memory-backed dedup so you don't process the same webhook twice
• Storage Abstraction: MemoryStorage (dev/test) and RedisStorage (production)
• Full Webhook Support: Parse, verify HMAC signatures, handle all message types

Installation

# Core library
pip install ai-prishtina-whatsapp-mcp

# With MCP server support (FastMCP)
pip install "ai-prishtina-whatsapp-mcp[mcp]"

# With AI provider dependencies
pip install "ai-prishtina-whatsapp-mcp[ai]"

# The whole enchilada
pip install "ai-prishtina-whatsapp-mcp[mcp,ai,storage,server]"

Requirements: Python 3.11+, Redis 6.0+ (optional, for production storage)

---

Quick Start

import asyncio
from ai_prishtina_whatsapp_mcp import WhatsAppClient, Settings
from ai_prishtina_whatsapp_mcp.config.settings import ProviderSettings

async def main():
    settings = Settings(
        whatsapp_access_token="your_token",
        whatsapp_phone_number_id="your_phone_id",
        providers=[
            ProviderSettings.openai(api_key="sk-..."),
            ProviderSettings.ollama(),  # For when the internet breaks
        ],
    )

    async with WhatsAppClient(settings=settings) as client:
        result = await client.send_text("[phone]", "Mirë se vini! 👋")

        if result.is_ok():
            print(f"Sent: {result.unwrap().id}")
        else:
            print(f"[{result.error.code.name}]: {result.error.message}")

asyncio.run(main())

---

Core Concepts

Result Monad — No More Try/Catch Roulette

Every operation returns Ok(value) or Err(DomainError). Your code stays clean and composable:

result = await client.send_text("[phone]", "Hello!")

if result.is_ok():
    msg = result.unwrap()
else:
    err = result.error
    print(f"[{err.code.name}] {err.message}")

# Or use expect() — raises RuntimeError if Err, returns value if Ok
msg = result.expect("send_text failed")

# Or chain
msg_id = result.map(lambda m: m.id).unwrap_or("unknown")

Rate Limiter

Token-bucket rate limiting built into the client — Meta's quota enforcement becomes your problem only if you ignore it:

await client.set_rate_limit(messages_per_second=5.0)
await client.send_text("[phone]", "message 1")
await client.send_text("[phone]", "message 2")  # automatically waits if needed

Bulk Messaging

results = await client.send_bulk_messages(
    recipients=["[phone]", "[phone]", "[phone]"],
    text="Big news from Prishtina! 🦅",
    delay_between=0.1,
)

for phone, result in results:
    print(f"{'✅' if result.is_ok() else '❌'} {phone}")

Conversation Buffer

Sliding-window context memory per phone number, ready to pass to any LLM:

from ai_prishtina_whatsapp_mcp import ConversationBuffer

buffer = ConversationBuffer(max_messages=20)
buffer.add("[phone]", role="user", content="What's the weather?")
buffer.add("[phone]", role="assistant", content="Sunny with a chance of 🦅")

history = buffer.get_history("[phone]")
# Feed directly to your LLM

Message Pipeline (Middleware)

Composable async middleware chains for webhook processing:

from ai_prishtina_whatsapp_mcp import MessagePipeline

pipeline = MessagePipeline()

@pipeline.use
async def log_middleware(event, ctx, next):
    print(f"Incoming: {event['type']}")
    await next(event, ctx)

@pipeline.on_text
async def handle_text(event, ctx):
    text = event.get("text", "")
    result = await client.analyze_sentiment(text)
    if result.is_ok():
        print(f"Sentiment: {result.unwrap().content}")

await pipeline.dispatch(webhook_event)

Message Deduplication

WhatsApp sometimes delivers the same webhook twice. We handle that:

from ai_prishtina_whatsapp_mcp.effects.storage import MemoryStorage

storage = MemoryStorage()
await storage.initialize()

if not await client.is_duplicate_message("wamid.abc123", storage, ttl=86400):
    await process_message(event)

---

AI Providers

9 providers, registered once, with automatic fallback:

from ai_prishtina_whatsapp_mcp.config.settings import ProviderSettings

settings = Settings(
    providers=[
        ProviderSettings.openai(api_key="sk-..."),
        ProviderSettings.anthropic(api_key="ant-..."),
        ProviderSettings.ollama(base_url="http://localhost:11434", model="llama3"),
        ProviderSettings.google(api_key="AIza..."),
        ProviderSettings.whisper(model_size="base"),
    ],
)

Provider	Text	Chat	Sentiment	Translation	Embeddings	Transcription
OpenAI	✅	✅	✅	✅	✅	—
Anthropic	✅	✅	✅	✅	—	—
Google Gemini	✅	✅	✅	✅	✅	—
AWS Bedrock	✅	✅	✅	—	✅	—
Azure OpenAI	✅	✅	✅	✅	✅	—
Ollama (local)	✅	✅	✅	—	✅	—
HuggingFace	✅	—	✅	✅	✅	—
Whisper (local)	—	—	—	—	—	✅
spaCy (local)	—	—	✅	—	✅	—

---

Storage

from ai_prishtina_whatsapp_mcp.effects.storage import MemoryStorage, RedisStorage

# Development / testing
storage = MemoryStorage()

# Production
storage = RedisStorage(redis_url="redis://localhost:6379")

await storage.initialize()

await storage.store("session:[phone]", {"intent": "order_pizza"}, ttl=3600)
data = await storage.retrieve("session:[phone]")
keys = await storage.list_keys("session:*")
exists = await storage.exists("session:[phone]")

---

Webhook Handling

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request):
    raw_body = await request.body()
    signature = request.headers.get("X-Hub-Signature-256", "")

    if not client.verify_webhook_signature(raw_body, signature):
        return {"status": "nice try"}

    events = client.parse_webhook(raw_body)
    for event in events:
        await pipeline.dispatch(event)

    return {"status": "ok"}

---

MCP Server

Run as CLI

# stdio transport (Claude Desktop, Windsurf, Cursor)
WHATSAPP_ACCESS_TOKEN=... WHATSAPP_PHONE_NUMBER_ID=... whatsapp-mcp

# HTTP transport
whatsapp-mcp --transport streamable-http --port 8000

Programmatic use

from ai_prishtina_whatsapp_mcp.mcp_server import create_server

server = create_server()
server.run()

---

Environment Variables

Variable	Description
WHATSAPP_ACCESS_TOKEN	Meta API access token (required)
WHATSAPP_PHONE_NUMBER_ID	WhatsApp phone number ID (required)
WHATSAPP_BUSINESS_ACCOUNT_ID	Business account ID
WHATSAPP_WEBHOOK_SECRET	Webhook HMAC-SHA256 secret
REDIS_URL	Redis connection URL (default: redis://localhost:6379)
OPENAI_API_KEY	OpenAI API key
ANTHROPIC_API_KEY	Anthropic API key
GOOGLE_API_KEY	Google Gemini API key
OLLAMA_BASE_URL	Ollama server URL

---

Architecture

ai_prishtina_whatsapp_mcp/
├── api.py                      # Low-level HTTP client (WhatsApp Business API)
├── mcp_server.py               # FastMCP server — 16 WhatsApp tools
├── config/
│   └── settings.py             # Settings, ProviderSettings, StorageSettings
├── core/
│   ├── domain.py               # Immutable domain models (frozen dataclasses)
│   ├── types.py                # Result<T,E>, Ok, Err, ErrorCode, DomainError
│   ├── whatsapp_client.py      # High-level client (send, bulk, dedup, rate limit)
│   ├── conversation_buffer.py  # Sliding-window context per phone number
│   └── pipeline.py             # Async middleware chain for webhooks
├── providers/
│   ├── base.py                 # AIProvider ABC, ProviderCapability enum
│   ├── registry.py             # AIProviderRegistry with health + fallback
│   ├── cloud/                  # OpenAI, Anthropic, Google, AWS, Azure
│   └── local/                  # Ollama, Whisper, HuggingFace, spaCy
├── effects/
│   └── storage.py              # MemoryStorage, RedisStorage
└── utils/
    ├── rate_limiter.py         # Async token-bucket rate limiter
    └── retry.py                # Exponential backoff retry utility

---

Testing

pip install -e ".[mcp,ai]"
pip install -r requirements-dev.txt

pytest tests/ -v
pytest tests/ --cov=ai_prishtina_whatsapp_mcp --cov-report=term-missing

---

Contributing

git clone https://github.com/albanmaxhuni/ai-prishtina-whatsapp-mcp
cd ai-prishtina-whatsapp-mcp
pip install -e ".[mcp,ai]"
pip install -r requirements-dev.txt
pytest tests/

PRs welcome. The library is from Prishtina 🦅 — we ship things.

---

License

AGPL-3.0+ OR Commercial — open source with copyleft, or contact us for commercial licensing.

---

Support

• Issues: GitHub Issues
• Email: [e-mail]

---

Built with passion (and plenty of coffee) by AI Prishtina. Go Eagles. 🦅