WhatsApp MCP Server

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)

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")

• src/whatsapp_mcp/modules/message.py:20-66 (helper)

  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)}")