WhatsApp MCP server

whatsapp_mcp_server.py

# whatsapp_server.py

import os
from typing import Annotated

from pydantic import Field, BeforeValidator
from pydantic_settings import BaseSettings, SettingsConfigDict
from twilio.rest import Client as TwilioClient
from twilio.base.exceptions import TwilioRestException

from fastmcp import FastMCP

# --- Configuration ---
# Use pydantic-settings to load credentials securely from environment or .env file
class TwilioSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="TWILIO_", env_file=".env")

    account_sid: str = Field(..., description="Twilio Account SID")
    auth_token: str = Field(..., description="Twilio Auth Token")
    whatsapp_from_number: Annotated[
        str,
        BeforeValidator(lambda v: f"whatsapp:{v}" if not v.startswith("whatsapp:") else v),
        Field(description="Twilio WhatsApp 'From' number (e.g., +14155238886)"),
    ]

# --- FastMCP Server Setup ---
mcp = FastMCP(
    name="Twilio WhatsApp Sender",
    instructions="Provides a tool to send WhatsApp messages via Twilio.",
    # Declare dependencies for `fastmcp install`
    dependencies=["twilio", "pydantic-settings", "python-dotenv"],
)

# Load settings
try:
    settings = TwilioSettings() # type: ignore
    twilio_client = TwilioClient(settings.account_sid, settings.auth_token)
except Exception as e:
    print(f"Error loading Twilio settings or initializing client: {e}")
    print("Please ensure TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, and TWILIO_WHATSAPP_FROM_NUMBER are set in your environment or .env file.")
    # Exit or handle gracefully if settings are essential for server startup
    # For this example, we'll let it proceed but the tool will fail later.
    settings = None # type: ignore
    twilio_client = None # type: ignore


def _ensure_whatsapp_prefix(v: str) -> str:
    """Validator to ensure the 'to' number starts with 'whatsapp:'."""
    if isinstance(v, str) and not v.startswith("whatsapp:"):
        # Attempt basic E.164 format check before prefixing
        if v.startswith("+") and v[1:].isdigit():
            return f"whatsapp:{v}"
        else:
            raise ValueError(f"Invalid phone number format: '{v}'. Must start with '+' and country code.")
    return v

# --- Tool Definition ---
@mcp.tool()
def send_whatsapp_message(
    to_number: Annotated[
        str,
        BeforeValidator(_ensure_whatsapp_prefix),
        Field(description="Recipient's WhatsApp number including country code (e.g., +15551234567)")
    ],
    message_body: Annotated[str, Field(description="The text content of the message to send.")]
) -> str:
    """
    Sends a WhatsApp message to the specified recipient using Twilio.
    Requires recipient number in E.164 format (e.g., +15551234567).
    """
    if not settings or not twilio_client:
        raise ValueError("Twilio settings are not configured correctly.")

    try:
        print(f"Attempting to send message to: {to_number}") # Debug print
        print(f"From number: {settings.whatsapp_from_number}") # Debug print

        message = twilio_client.messages.create(
            from_=settings.whatsapp_from_number,
            body=message_body,
            to=to_number # Already prefixed by validator
        )
        print(f"Twilio message SID: {message.sid}, Status: {message.status}") # Debug print
        return f"WhatsApp message sent successfully! SID: {message.sid}"

    except TwilioRestException as e:
        print(f"Twilio API error: {e}") # Debug print
        # Raise a clearer exception for MCP
        raise ValueError(f"Failed to send WhatsApp message via Twilio: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}") # Debug print
        # Catch other potential errors
        raise ValueError(f"An unexpected error occurred: {e}")

# --- Main Execution Block ---
if __name__ == "__main__":
    print("Starting Twilio WhatsApp MCP server...")
    if not settings or not twilio_client:
         print("\nWARNING: Twilio client not initialized due to missing settings.\n"
               "The 'send_whatsapp_message' tool will fail until configured.\n")
    mcp.run() # Runs the server on stdio by default