PyWA MCP Server

Instructions

Implementation Reference

• tools/messaging.py:290-320 (handler)

  The request_location tool handler that requests a user's location via WhatsApp. It takes parameters 'to' (phone number/WhatsApp ID), 'text' (message text), and optional 'reply_to_message_id'. Uses wa_client.request_location() to send the request and returns a dictionary with success status and message ID.
  @mcp.tool() async def request_location( to: str, text: str, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Request user's location. Args: to: Phone number or WhatsApp ID text: Message text asking for location reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.request_location( to=to, text=text, reply_to_message_id=reply_to_message_id, ) logger.info(f"Location request sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to request location: {str(e)}") return {"success": False, "error": str(e)}
• tools/messaging.py:11-320 (registration)

  The register_messaging_tools function that registers all messaging tools including request_location. The request_location tool is registered with the @mcp.tool() decorator on line 290 within this function.
  def register_messaging_tools(mcp, wa_client: WhatsApp): """Register all messaging-related tools.""" @mcp.tool() async def send_message( to: str, text: str, header: Optional[str] = None, footer: Optional[str] = None, *, preview_url: bool = False, reply_to_message_id: Optional[str] = None, ) -> dict: """ Send a text message to a WhatsApp user. Args: to: Phone number (with country code) or WhatsApp ID text: The text message content header: Optional header text (for interactive messages) footer: Optional footer text (for interactive messages) preview_url: Whether to show URL previews (default False) reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.send_message( to=to, text=text, header=header, footer=footer, preview_url=preview_url, reply_to_message_id=reply_to_message_id, ) logger.info(f"Message sent to {to}") # Extract just the message ID if result is a complex object message_id = getattr(result, 'id', str(result)) if result else None return { "success": True, "message_id": message_id, "to": to, } except Exception as e: logger.error(f"Failed to send message: {str(e)}") return {"success": False, "error": str(e)} @mcp.tool() async def send_image( to: str, image: str, caption: Optional[str] = None, footer: Optional[str] = None, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Send an image message. Args: to: Phone number or WhatsApp ID image: Image URL or media ID caption: Optional image caption footer: Optional footer text reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.send_image( to=to, image=image, caption=caption, footer=footer, reply_to_message_id=reply_to_message_id, ) logger.info(f"Image sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to send image: {str(e)}") return {"success": False, "error": str(e)} @mcp.tool() async def send_video( to: str, video: str, caption: Optional[str] = None, footer: Optional[str] = None, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Send a video message. Args: to: Phone number or WhatsApp ID video: Video URL or media ID caption: Optional video caption footer: Optional footer text reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.send_video( to=to, video=video, caption=caption, footer=footer, reply_to_message_id=reply_to_message_id, ) logger.info(f"Video sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to send video: {str(e)}") return {"success": False, "error": str(e)} @mcp.tool() async def send_document( to: str, document: str, filename: Optional[str] = None, caption: Optional[str] = None, footer: Optional[str] = None, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Send a document message. Args: to: Phone number or WhatsApp ID document: Document URL or media ID filename: Optional filename for the document caption: Optional document caption footer: Optional footer text reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.send_document( to=to, document=document, filename=filename, caption=caption, footer=footer, reply_to_message_id=reply_to_message_id, ) logger.info(f"Document sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to send document: {str(e)}") return {"success": False, "error": str(e)} @mcp.tool() async def send_audio( to: str, audio: str, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Send an audio message. Args: to: Phone number or WhatsApp ID audio: Audio URL or media ID reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.send_audio( to=to, audio=audio, reply_to_message_id=reply_to_message_id, ) logger.info(f"Audio sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to send audio: {str(e)}") return {"success": False, "error": str(e)} @mcp.tool() async def send_sticker( to: str, sticker: str, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Send a sticker message. Args: to: Phone number or WhatsApp ID sticker: Sticker URL or media ID (must be webp format) reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.send_sticker( to=to, sticker=sticker, reply_to_message_id=reply_to_message_id, ) logger.info(f"Sticker sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to send sticker: {str(e)}") return {"success": False, "error": str(e)} @mcp.tool() async def send_location( to: str, latitude: float, longitude: float, name: Optional[str] = None, address: Optional[str] = None, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Send a location message. Args: to: Phone number or WhatsApp ID latitude: Latitude of the location longitude: Longitude of the location name: Optional location name address: Optional location address reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.send_location( to=to, latitude=latitude, longitude=longitude, name=name, address=address, reply_to_message_id=reply_to_message_id, ) logger.info(f"Location sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to send location: {str(e)}") return {"success": False, "error": str(e)} @mcp.tool() async def request_location( to: str, text: str, *, reply_to_message_id: Optional[str] = None, ) -> dict: """ Request user's location. Args: to: Phone number or WhatsApp ID text: Message text asking for location reply_to_message_id: Message ID to reply to Returns: Dictionary with success status and message ID """ try: result = wa_client.request_location( to=to, text=text, reply_to_message_id=reply_to_message_id, ) logger.info(f"Location request sent to {to}") message_id = getattr(result, 'id', str(result)) if result else None return {"success": True, "message_id": message_id} except Exception as e: logger.error(f"Failed to request location: {str(e)}") return {"success": False, "error": str(e)}
• tools/__init__.py:14-18 (registration)

  The register_all_tools function that registers all available tools with the MCP server, including calling register_messaging_tools which contains the request_location tool.
  def register_all_tools(mcp, wa_client): """Register all available tools with the MCP server.""" register_messaging_tools(mcp, wa_client) register_interactive_tools(mcp, wa_client) register_template_tools(mcp, wa_client)
• server.py:68-68 (registration)

  The server initialization where register_all_tools is called to register all tools (including request_location) with the MCP server instance.
  register_all_tools(mcp, client)