bulk_message

Instructions

Implementation Reference

• src/yaparai/tools/crm.py:121-158 (handler)

  The bulk_message async function — the actual tool handler. It accepts a message, optional customer_ids, tag, platform, media_urls, and org_id. It builds a payload dict and delegates to client.crm_bulk_message().

  async def bulk_message(
      message: str,
      customer_ids: list[str] | None = None,
      tag: str | None = None,
      platform: Literal["instagram", "facebook", "whatsapp"] | None = None,
      media_urls: list[str] | None = None,
      org_id: str | None = None,
  ) -> dict:
      """
      Send a message to multiple customers at once.
  
      Broadcast promotions, announcements, or updates to a list of
      customers via their respective social platforms. You can target by
      specific IDs, a customer tag, or a platform.
  
      Args:
          message: Message text to send
          customer_ids: List of specific customer IDs to message
          tag: Send to all customers with this tag (e.g., "vip", "returning")
          platform: Send only to customers from this platform
          media_urls: Optional list of image/video URLs to attach
          org_id: Organization ID (uses YAPARAI_ORG_ID env var if not provided)
  
      Returns:
          Dict with sent count, failed count, and error details.
      """
      oid = resolve_org_id(org_id)
      client = YaparAIClient()
      payload: dict = {"message": message}
      if customer_ids:
          payload["customer_ids"] = customer_ids
      if tag:
          payload["tag"] = tag
      if platform:
          payload["platform"] = platform
      if media_urls:
          payload["media_urls"] = media_urls
      return await client.crm_bulk_message(oid, payload)

• src/yaparai/server.py:89-89 (registration)

  Import of bulk_message from yaparai.tools.crm into the server module.

  bulk_message,

• src/yaparai/server.py:173-173 (registration)

  Registration of bulk_message as an MCP tool via mcp.tool(bulk_message).

  mcp.tool(bulk_message)

• src/yaparai/client.py:303-307 (helper)

  The crm_bulk_message method on YaparAIClient that makes the actual HTTP POST request to /api/enterprise/orgs/{org_id}/crm/bulk-message.

  async def crm_bulk_message(self, org_id: str, payload: dict) -> dict:
      """Send bulk message to customers."""
      return await self._request(
          "POST", f"/api/enterprise/orgs/{org_id}/crm/bulk-message", json=payload
      )

• tests/test_crm.py:5-41 (helper)

  Tests for bulk_message: test_bulk_message_by_ids (line 9), test_bulk_message_by_tag (line 21), and test_bulk_message_by_platform (line 33).

  from yaparai.tools.crm import bulk_message
  
  
  @pytest.mark.asyncio
  async def test_bulk_message_by_ids():
      with patch("yaparai.tools.crm.YaparAIClient") as M, \
           patch("yaparai.tools.crm.resolve_org_id", return_value="org1"):
          inst = M.return_value
          inst.crm_bulk_message = AsyncMock(return_value={"sent": 3, "failed": 0})
          await bulk_message(message="Hi!", customer_ids=["c1", "c2"])
          payload = inst.crm_bulk_message.call_args[0][1]
          assert payload["customer_ids"] == ["c1", "c2"]
          assert "tag" not in payload
  
  
  @pytest.mark.asyncio
  async def test_bulk_message_by_tag():
      with patch("yaparai.tools.crm.YaparAIClient") as M, \
           patch("yaparai.tools.crm.resolve_org_id", return_value="org1"):
          inst = M.return_value
          inst.crm_bulk_message = AsyncMock(return_value={"sent": 5, "failed": 0})
          await bulk_message(message="VIP offer!", tag="vip")
          payload = inst.crm_bulk_message.call_args[0][1]
          assert payload["tag"] == "vip"
          assert "customer_ids" not in payload
  
  
  @pytest.mark.asyncio
  async def test_bulk_message_by_platform():
      with patch("yaparai.tools.crm.YaparAIClient") as M, \
           patch("yaparai.tools.crm.resolve_org_id", return_value="org1"):
          inst = M.return_value
          inst.crm_bulk_message = AsyncMock(return_value={"sent": 2, "failed": 0})
          await bulk_message(message="promo", platform="instagram")
          payload = inst.crm_bulk_message.call_args[0][1]
          assert payload["platform"] == "instagram"