#!/usr/bin/env python3
"""
MCP Server for Uazapi WhatsApp API v2.0
This server provides tools to interact with Uazapi WhatsApp API, enabling
message sending, contact management, group operations, and more.
"""
from typing import Optional, List, Dict, Any
from enum import Enum
import os
import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict
from mcp.server.fastmcp import FastMCP, Context
# Initialize the MCP server
mcp = FastMCP("uazapi_mcp")
# Constants
CHARACTER_LIMIT = 25000 # Maximum response size in characters
DEFAULT_TIMEOUT = 30.0 # Default API request timeout in seconds
# Environment variables for authentication
# Users should set: UAZAPI_API_KEY and UAZAPI_INSTANCE_ID
API_KEY = os.getenv("UAZAPI_API_KEY", "")
INSTANCE_ID = os.getenv("UAZAPI_INSTANCE_ID", "")
API_BASE_URL = f"https://api.uazapi.com/instances/{INSTANCE_ID}" if INSTANCE_ID else ""
# Enums
class ResponseFormat(str, Enum):
"""Output format for tool responses."""
MARKDOWN = "markdown"
JSON = "json"
class MessageType(str, Enum):
"""Types of messages that can be sent via Uazapi."""
TEXT = "text"
IMAGE = "image"
VIDEO = "video"
AUDIO = "audio"
DOCUMENT = "document"
LOCATION = "location"
CONTACT = "contact"
STICKER = "sticker"
# ===== SHARED UTILITY FUNCTIONS =====
def _validate_auth() -> bool:
"""Validate that required authentication credentials are set."""
return bool(API_KEY and INSTANCE_ID and API_BASE_URL)
def _get_auth_headers() -> Dict[str, str]:
"""Get authentication headers for API requests."""
return {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
async def _make_api_request(
endpoint: str,
method: str = "GET",
params: Optional[Dict[str, Any]] = None,
json_data: Optional[Dict[str, Any]] = None,
timeout: float = DEFAULT_TIMEOUT
) -> Dict[str, Any]:
"""
Reusable function for all Uazapi API calls.
Args:
endpoint: API endpoint path (e.g., "messages/send")
method: HTTP method (GET, POST, PUT, DELETE)
params: Query parameters
json_data: JSON body for POST/PUT requests
timeout: Request timeout in seconds
Returns:
JSON response from the API
Raises:
httpx.HTTPStatusError: On HTTP errors
httpx.TimeoutException: On request timeout
"""
if not _validate_auth():
raise ValueError(
"Authentication not configured. Please set UAZAPI_API_KEY and "
"UAZAPI_INSTANCE_ID environment variables."
)
url = f"{API_BASE_URL}/{endpoint}"
headers = _get_auth_headers()
async with httpx.AsyncClient() as client:
response = await client.request(
method=method,
url=url,
headers=headers,
params=params,
json=json_data,
timeout=timeout
)
response.raise_for_status()
return response.json()
def _handle_api_error(e: Exception) -> str:
"""
Consistent error formatting across all tools.
Provides clear, actionable error messages to guide users.
"""
if isinstance(e, ValueError):
return f"Error: {str(e)}"
elif isinstance(e, httpx.HTTPStatusError):
status_code = e.response.status_code
if status_code == 400:
return (
"Error: Bad request. Please check your input parameters are correct. "
f"Details: {e.response.text}"
)
elif status_code == 401:
return (
"Error: Unauthorized. Please check your UAZAPI_API_KEY is correct and valid."
)
elif status_code == 403:
return (
"Error: Permission denied. Your API key doesn't have access to this resource."
)
elif status_code == 404:
return (
"Error: Resource not found. Please verify the ID or endpoint exists."
)
elif status_code == 429:
return (
"Error: Rate limit exceeded. Please wait a moment before making more requests."
)
elif status_code >= 500:
return (
f"Error: Uazapi server error (status {status_code}). "
"The service may be temporarily unavailable. Please try again later."
)
return f"Error: API request failed with status {status_code}: {e.response.text}"
elif isinstance(e, httpx.TimeoutException):
return (
"Error: Request timed out. The API took too long to respond. "
"Please try again or check your network connection."
)
elif isinstance(e, httpx.RequestError):
return f"Error: Network error - {str(e)}. Please check your internet connection."
return f"Error: Unexpected error occurred: {type(e).__name__} - {str(e)}"
def _format_phone_number(phone: str) -> str:
"""
Format phone number for WhatsApp/Uazapi.
Removes special characters and ensures proper format.
Expected format: Country code + number (e.g., "5511999999999")
"""
# Remove common separators
cleaned = phone.replace("-", "").replace(" ", "").replace("(", "").replace(")", "")
# Remove + prefix if present
if cleaned.startswith("+"):
cleaned = cleaned[1:]
return cleaned
def _truncate_response(content: str, metadata: Optional[Dict[str, Any]] = None) -> str:
"""
Truncate response if it exceeds CHARACTER_LIMIT.
Args:
content: The response content to check
metadata: Optional metadata about truncation
Returns:
Original or truncated content with truncation message
"""
if len(content) <= CHARACTER_LIMIT:
return content
# Truncate and add clear message
truncated = content[:CHARACTER_LIMIT]
truncation_msg = (
f"\n\n--- RESPONSE TRUNCATED ---\n"
f"Original size: {len(content)} characters\n"
f"Showing first {CHARACTER_LIMIT} characters.\n"
)
if metadata:
if "total_items" in metadata and "shown_items" in metadata:
truncation_msg += (
f"Showing {metadata['shown_items']} of {metadata['total_items']} items. "
f"Use pagination parameters (limit, offset) to see more.\n"
)
return truncated + truncation_msg
# ===== PYDANTIC MODELS FOR INPUT VALIDATION =====
class SendTextMessageInput(BaseModel):
"""Input model for sending text messages via WhatsApp."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
phone: str = Field(
...,
description="Recipient's phone number with country code (e.g., '5511999999999', '+55 11 99999-9999')",
min_length=10,
max_length=20
)
message: str = Field(
...,
description="Text message content to send",
min_length=1,
max_length=4096
)
@field_validator('phone')
@classmethod
def validate_phone(cls, v: str) -> str:
"""Validate and format phone number."""
formatted = _format_phone_number(v)
if not formatted.isdigit():
raise ValueError("Phone number must contain only digits after formatting")
if len(formatted) < 10:
raise ValueError("Phone number too short (must be at least 10 digits with country code)")
return formatted
class SendMediaMessageInput(BaseModel):
"""Input model for sending media messages (image, video, audio, document)."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
phone: str = Field(
...,
description="Recipient's phone number with country code (e.g., '5511999999999')",
min_length=10,
max_length=20
)
media_url: str = Field(
...,
description="Public URL of the media file to send (must be HTTPS)",
min_length=10,
max_length=2048
)
media_type: MessageType = Field(
...,
description="Type of media: image, video, audio, or document"
)
caption: Optional[str] = Field(
default=None,
description="Optional caption for the media",
max_length=1024
)
filename: Optional[str] = Field(
default=None,
description="Optional filename for documents",
max_length=255
)
@field_validator('phone')
@classmethod
def validate_phone(cls, v: str) -> str:
formatted = _format_phone_number(v)
if not formatted.isdigit():
raise ValueError("Phone number must contain only digits")
return formatted
@field_validator('media_url')
@classmethod
def validate_url(cls, v: str) -> str:
if not v.startswith(('http://', 'https://')):
raise ValueError("Media URL must start with http:// or https://")
return v
@field_validator('media_type')
@classmethod
def validate_media_type(cls, v: MessageType) -> MessageType:
allowed = [MessageType.IMAGE, MessageType.VIDEO, MessageType.AUDIO, MessageType.DOCUMENT]
if v not in allowed:
raise ValueError(f"Media type must be one of: {', '.join([t.value for t in allowed])}")
return v
class GetContactsInput(BaseModel):
"""Input model for retrieving contacts."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True
)
limit: Optional[int] = Field(
default=20,
description="Maximum number of contacts to return",
ge=1,
le=100
)
offset: Optional[int] = Field(
default=0,
description="Number of contacts to skip for pagination",
ge=0
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
class GetChatsInput(BaseModel):
"""Input model for retrieving chats/conversations."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True
)
limit: Optional[int] = Field(
default=20,
description="Maximum number of chats to return",
ge=1,
le=100
)
offset: Optional[int] = Field(
default=0,
description="Number of chats to skip for pagination",
ge=0
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
# ===== TOOL IMPLEMENTATIONS =====
@mcp.tool(
name="uazapi_send_text_message",
annotations={
"title": "Send WhatsApp Text Message",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def uazapi_send_text_message(params: SendTextMessageInput) -> str:
"""
Send a text message to a WhatsApp contact via Uazapi API.
This tool sends simple text messages to any WhatsApp number. It validates
phone numbers, formats them correctly, and handles the API communication.
Args:
params (SendTextMessageInput): Validated input parameters containing:
- phone (str): Recipient's phone number with country code (e.g., "5511999999999")
- message (str): Text content to send (1-4096 characters)
Returns:
str: Success message with message ID or error description
Success response format:
"Message sent successfully!
- Recipient: +5511999999999
- Message ID: wamid.xxxxxxxxxxxxx
- Status: sent"
Error response format:
"Error: <detailed error message with guidance>"
Examples:
- Use when: "Send 'Hello World' to +55 11 99999-9999"
- Use when: "Message the customer at 5511999999999 saying their order is ready"
- Don't use when: You need to send images/files (use uazapi_send_media_message)
- Don't use when: You need to send to multiple people (call this tool multiple times)
Error Handling:
- Returns "Error: Authentication not configured" if API credentials are missing
- Returns "Error: Bad request" if phone number format is invalid
- Returns "Error: Rate limit exceeded" if too many requests (429 status)
- Pydantic handles input validation (length, format, required fields)
"""
try:
# Prepare request payload
payload = {
"phone": params.phone,
"message": params.message
}
# Make API request
response = await _make_api_request(
endpoint="messages/send/text",
method="POST",
json_data=payload
)
# Format success response
message_id = response.get("id", "unknown")
status = response.get("status", "unknown")
return (
f"Message sent successfully!\n"
f"- Recipient: +{params.phone}\n"
f"- Message ID: {message_id}\n"
f"- Status: {status}"
)
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="uazapi_send_media_message",
annotations={
"title": "Send WhatsApp Media Message",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def uazapi_send_media_message(params: SendMediaMessageInput) -> str:
"""
Send a media message (image, video, audio, or document) to a WhatsApp contact.
This tool sends media files via URL to WhatsApp contacts. The media file must
be publicly accessible via HTTPS. Supports images, videos, audio files, and documents.
Args:
params (SendMediaMessageInput): Validated input parameters containing:
- phone (str): Recipient's phone number with country code
- media_url (str): Public HTTPS URL of the media file
- media_type (MessageType): Type of media (image/video/audio/document)
- caption (Optional[str]): Optional caption text for the media
- filename (Optional[str]): Optional filename for documents
Returns:
str: Success message with message ID or error description
Success response format:
"Media message sent successfully!
- Recipient: +5511999999999
- Media Type: image
- Message ID: wamid.xxxxxxxxxxxxx"
Examples:
- Use when: "Send the image at https://example.com/photo.jpg to +5511999999999"
- Use when: "Share this PDF https://site.com/doc.pdf with caption 'Report' to contact"
- Don't use when: Sending plain text (use uazapi_send_text_message)
- Don't use when: Media URL is not publicly accessible
Error Handling:
- Validates media_url is HTTPS
- Validates media_type is appropriate for the operation
- Returns clear errors for authentication, rate limits, or invalid URLs
"""
try:
# Prepare request payload based on media type
payload = {
"phone": params.phone,
"mediaUrl": params.media_url,
"mediaType": params.media_type.value
}
if params.caption:
payload["caption"] = params.caption
if params.filename and params.media_type == MessageType.DOCUMENT:
payload["filename"] = params.filename
# Make API request
response = await _make_api_request(
endpoint="messages/send/media",
method="POST",
json_data=payload
)
# Format success response
message_id = response.get("id", "unknown")
result = (
f"Media message sent successfully!\n"
f"- Recipient: +{params.phone}\n"
f"- Media Type: {params.media_type.value}\n"
f"- Message ID: {message_id}"
)
if params.caption:
result += f"\n- Caption: {params.caption}"
return result
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="uazapi_get_contacts",
annotations={
"title": "Get WhatsApp Contacts",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def uazapi_get_contacts(params: GetContactsInput) -> str:
"""
Retrieve WhatsApp contacts from the Uazapi instance.
This tool fetches the list of contacts saved in your WhatsApp instance.
Supports pagination for large contact lists and multiple output formats.
Args:
params (GetContactsInput): Validated input parameters containing:
- limit (Optional[int]): Max contacts to return (1-100, default: 20)
- offset (Optional[int]): Number to skip for pagination (default: 0)
- response_format (ResponseFormat): Output format (markdown/json)
Returns:
str: Formatted list of contacts or error message
Markdown format example:
"# WhatsApp Contacts
Found 45 contacts (showing 20)
## John Doe (+5511999999999)
- Name: John Doe
- Phone: +5511999999999
- Status: Available
## Jane Smith (+5511888888888)
..."
JSON format example:
{
"total": 45,
"count": 20,
"offset": 0,
"has_more": true,
"next_offset": 20,
"contacts": [
{
"id": "5511999999999@c.us",
"name": "John Doe",
"phone": "5511999999999",
"pushname": "John",
"isMyContact": true
}
]
}
Examples:
- Use when: "Show me my WhatsApp contacts"
- Use when: "List the first 50 contacts"
- Use when: "Get contacts starting from position 20"
- Don't use when: Searching for a specific contact by name (filter results yourself)
Error Handling:
- Handles pagination automatically
- Truncates if response exceeds CHARACTER_LIMIT
- Returns clear message if no contacts found
"""
try:
# Make API request with pagination
response = await _make_api_request(
endpoint="contacts",
method="GET",
params={
"limit": params.limit,
"offset": params.offset
}
)
contacts = response.get("contacts", [])
total = response.get("total", len(contacts))
if not contacts:
return "No contacts found in your WhatsApp instance."
# Format response based on requested format
if params.response_format == ResponseFormat.MARKDOWN:
lines = ["# WhatsApp Contacts\n"]
lines.append(f"Found {total} contacts (showing {len(contacts)})\n")
for contact in contacts:
name = contact.get("name", contact.get("pushname", "Unknown"))
phone = contact.get("phone", contact.get("id", "").split("@")[0])
lines.append(f"## {name} (+{phone})")
lines.append(f"- Name: {name}")
lines.append(f"- Phone: +{phone}")
if contact.get("isMyContact"):
lines.append("- Saved: Yes")
lines.append("")
content = "\n".join(lines)
# Check truncation
if len(content) > CHARACTER_LIMIT:
metadata = {"total_items": total, "shown_items": len(contacts)}
content = _truncate_response(content, metadata)
return content
else: # JSON format
import json
result = {
"total": total,
"count": len(contacts),
"offset": params.offset,
"has_more": total > params.offset + len(contacts),
"next_offset": params.offset + len(contacts) if total > params.offset + len(contacts) else None,
"contacts": contacts
}
return json.dumps(result, indent=2)
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="uazapi_get_chats",
annotations={
"title": "Get WhatsApp Chats",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def uazapi_get_chats(params: GetChatsInput) -> str:
"""
Retrieve recent WhatsApp chats/conversations from the Uazapi instance.
This tool fetches the list of recent conversations (individual and group chats)
from your WhatsApp instance. Supports pagination and multiple output formats.
Args:
params (GetChatsInput): Validated input parameters containing:
- limit (Optional[int]): Max chats to return (1-100, default: 20)
- offset (Optional[int]): Number to skip for pagination (default: 0)
- response_format (ResponseFormat): Output format (markdown/json)
Returns:
str: Formatted list of chats or error message
Markdown format shows:
- Chat name/contact
- Last message preview
- Unread count
- Timestamp
JSON format provides complete chat metadata
Examples:
- Use when: "Show me recent WhatsApp conversations"
- Use when: "List the last 30 chats"
- Use when: "Get chats with unread messages"
- Don't use when: You need full message history (use separate message retrieval)
Error Handling:
- Handles pagination automatically
- Truncates if response exceeds CHARACTER_LIMIT
- Returns clear message if no chats found
"""
try:
# Make API request with pagination
response = await _make_api_request(
endpoint="chats",
method="GET",
params={
"limit": params.limit,
"offset": params.offset
}
)
chats = response.get("chats", [])
total = response.get("total", len(chats))
if not chats:
return "No chats found in your WhatsApp instance."
# Format response based on requested format
if params.response_format == ResponseFormat.MARKDOWN:
lines = ["# WhatsApp Chats\n"]
lines.append(f"Found {total} chats (showing {len(chats)})\n")
for chat in chats:
name = chat.get("name", "Unknown")
chat_id = chat.get("id", "")
unread = chat.get("unreadCount", 0)
lines.append(f"## {name}")
lines.append(f"- Chat ID: {chat_id}")
if unread > 0:
lines.append(f"- **Unread Messages**: {unread}")
last_message = chat.get("lastMessage", {})
if last_message:
preview = last_message.get("body", "")[:100]
lines.append(f"- Last Message: {preview}...")
lines.append("")
content = "\n".join(lines)
# Check truncation
if len(content) > CHARACTER_LIMIT:
metadata = {"total_items": total, "shown_items": len(chats)}
content = _truncate_response(content, metadata)
return content
else: # JSON format
import json
result = {
"total": total,
"count": len(chats),
"offset": params.offset,
"has_more": total > params.offset + len(chats),
"next_offset": params.offset + len(chats) if total > params.offset + len(chats) else None,
"chats": chats
}
return json.dumps(result, indent=2)
except Exception as e:
return _handle_api_error(e)
# ===== SERVER ENTRY POINT =====
if __name__ == "__main__":
# Validate configuration on startup
if not _validate_auth():
print(
"WARNING: UAZAPI_API_KEY and UAZAPI_INSTANCE_ID environment variables not set.\n"
"The server will start but tools will fail until these are configured.\n"
"\n"
"Please set:\n"
" export UAZAPI_API_KEY='your_api_key_here'\n"
" export UAZAPI_INSTANCE_ID='your_instance_id_here'\n"
)
# Run the MCP server (stdio transport by default)
mcp.run()