WhatsApp MCP Server

Instructions

Implementation Reference

• src/whatsapp_mcp/server.py:103-123 (handler)

  MCP tool handler for 'create_group', decorated with @mcp.tool() which registers it with the server. Calls into the group module for execution.
  @mcp.tool() async def create_group(ctx: Context, group_name: str, participants: list[str]) -> str: """ Create a new WhatsApp group. Parameters: - group_name: Name of the group to create - participants: List of participant phone numbers """ try: if not auth.auth_manager.is_authenticated(): return "Error: No active session" group_result = await group.create_group( group_name=group_name, participants=participants, ) return json.dumps(group_result.model_dump()) except Exception as e: logger.error(f"Error creating group: {e}") return f"Error: {str(e)}"
• src/whatsapp_mcp/models.py:139-146 (schema)

  Pydantic model defining the input schema for the create_group tool.
  class CreateGroup(BaseModel): """Input schema for create_group tool.""" group_name: str = Field(..., description="Name of the group to create") participants: List[str] = Field( ..., description="List of participant phone numbers" )
• src/whatsapp_mcp/modules/group.py:16-91 (helper)

  Core helper function implementing the group creation logic, called by the MCP handler. Interacts with WhatsApp client to create the group.
  async def create_group(group_name: str, participants: List[str]) -> Group: """Create a new WhatsApp group.""" logger.info(f"Creating group {group_name} with {len(participants)} participants") 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") if len(participants) < 1: raise ValueError("Need at least one participant to create a group") try: # Format participant phone numbers correctly formatted_participants = [] for phone in participants: # Add @c.us suffix if not already present if not phone.endswith("@c.us"): formatted_phone = f"{phone}@c.us" else: formatted_phone = phone formatted_participants.append(formatted_phone) # Prepare the request data for group creation # Note: The exact API format may vary depending on the WhatsApp API being used group_data = {"group_name": group_name, "participants": formatted_participants} logger.debug(f"Creating group with data: {json.dumps(group_data)}") # Create the group via the WhatsApp API # The response format may vary depending on the API response = await asyncio.to_thread( whatsapp_client.client.create_group, group_data ) # Parse the response if not response or not response.get("success", False): error_msg = ( response.get("error", "Unknown error") if response else "No response" ) logger.error(f"Failed to create group: {error_msg}") raise ValueError(f"Failed to create group: {error_msg}") # Extract group information from response group_info = response.get("group", {}) group_id = group_info.get("id", f"{uuid.uuid4().hex[:12]}@g.us") # Create participant objects participant_objects = [] for i, phone in enumerate(participants): contact = Contact( id=formatted_participants[i], name=f"Participant {i + 1}", # We may not have names initially phone=phone, ) participant = Participant(id=contact.id, is_admin=False, contact=contact) participant_objects.append(participant) # Create the group object group = Group( id=group_id, name=group_name, description=group_info.get("description", ""), owner=group_info.get("owner", "me"), creation_time=datetime.now().isoformat(), participants=participant_objects, ) logger.info(f"Group created with ID {group_id}") return group except Exception as e: logger.error(f"Failed to create group: {e}") raise ValueError(f"Failed to create group: {str(e)}")
• src/whatsapp_mcp/server.py:103-103 (registration)

  The @mcp.tool() decorator registers the create_group function as an MCP tool.
  @mcp.tool()