send_message
Send messages directly to WhatsApp contacts using a recipient's phone number and message content. Optionally reply to specific messages with a reply_to parameter.
Instructions
Send a message to a chat.
Parameters:
- phone_number: The phone number of the recipient
- content: The content of the message to send
- reply_to: ID of the message to reply to (optional)Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| content | Yes | ||
| phone_number | Yes | ||
| reply_to | No |
Implementation Reference
- src/whatsapp_mcp/server.py:58-81 (handler)MCP tool handler for send_message. Decorated with @mcp.tool() for registration. Performs authentication check and delegates to the message module's send_message helper.
@mcp.tool() async def send_message( ctx: Context, phone_number: str, content: str, reply_to: Optional[str] = None ) -> str: """ Send a message to a chat. Parameters: - phone_number: The phone number of the recipient - content: The content of the message to send - reply_to: ID of the message to reply to (optional) """ try: if not auth.auth_manager.is_authenticated(): return "Error: No active session" result = await message.send_message( phone_number=phone_number, content=content, reply_to=reply_to ) return json.dumps(result) except Exception as e: logger.error(f"Error sending message: {e}") return f"Error: {str(e)}" - src/whatsapp_mcp/models.py:131-137 (schema)Pydantic BaseModel defining the input schema parameters for the send_message tool.
class SendMessage(BaseModel): """Input schema for send_message tool.""" phone_number: str = Field(..., description="The phone number of the recipient") content: str = Field(..., description="The content of the message to send") reply_to: str | None = Field(None, description="ID of the message to reply to") - Supporting function that implements the core logic of sending a message via the WhatsApp client's sendMessage API, including chat ID formatting and response handling.
async def send_message( phone_number: str, content: str, reply_to: Optional[str] = None ) -> dict: """Send a message to a chat.""" logger.info(f"Sending message to {phone_number}") whatsapp_client = auth_manager.get_client() if not whatsapp_client: raise ValueError("Session not found") if not whatsapp_client.client: raise ValueError("WhatsApp client not initialized") try: chat_id = _get_chat_id(phone_number) # Send the message via the WhatsApp API logger.debug(f"Sending message to {chat_id}: {content}") # Convert to asyncio to prevent blocking response = whatsapp_client.client.sending.sendMessage(chat_id, content) logger.info(f"Response code {response.code}: {response.data}") response_data = response.data message_id = "Not provided" # Try to extract message ID from the response if available if isinstance(response_data, dict): if response_data.get("idMessage"): message_id = response_data.get("idMessage") elif response_data.get("id"): message_id = response_data.get("id") result = { "message_id": message_id, "status": "sent", "timestamp": datetime.now().isoformat(), "response": response_data, } logger.info(f"Message sent with ID {message_id}") return result except Exception as e: logger.error(f"Failed to send message: {e}") raise ValueError(f"Failed to send message: {str(e)}")