create_audience

Instructions

Implementation Reference

• mcp_mailchimp/server.py:419-454 (handler)

  The actual handler function for the create_audience tool. Creates a new Mailchimp audience/list via POST /lists with name, sender email, company, permission reminder, and campaign defaults.

  @mcp.tool()
  async def create_audience(
      name: str,
      from_email: str,
      company: str,
      permission_reminder: str = "You signed up on our website.",
      from_name: str = "",
      country: str = "US",
  ) -> str:
      """Create a new audience/list. Requires name, sender email, and company name."""
      mc = get_client()
      body = {
          "name": name,
          "permission_reminder": permission_reminder,
          "email_type_option": True,
          "contact": {
              "company": company,
              "address1": "",
              "city": "",
              "state": "",
              "zip": "",
              "country": country,
          },
          "campaign_defaults": {
              "from_name": from_name or company,
              "from_email": from_email,
              "subject": "",
              "language": "en",
          },
      }
      a = await mc.post("/lists", json=body)
      return _fmt({
          "id": a["id"],
          "name": a.get("name", ""),
          "message": "Audience created.",
      })

• mcp_mailchimp/server.py:419-420 (registration)

  The @mcp.tool() decorator registers create_audience as an MCP tool.

  @mcp.tool()
  async def create_audience(

• mcp_mailchimp/server.py:27-38 (helper)

  get_client() creates/returns the singleton MailchimpClient used by create_audience to make the API call.

  def get_client() -> MailchimpClient:
      global _client
      if _client is None:
          api_key = os.environ.get("MAILCHIMP_API_KEY", "")
          if not api_key or "-" not in api_key:
              raise ValueError(
                  "MAILCHIMP_API_KEY environment variable required. "
                  "Format: xxxxxxxxxx-usXX "
                  "(get yours at https://mailchimp.com/account/api)"
              )
          _client = MailchimpClient(api_key)
      return _client

• mcp_mailchimp/server.py:41-43 (helper)

  _fmt() utility used by create_audience to format the response as indented JSON.

  def _fmt(data: Any) -> str:
      """Format response data as indented JSON string."""
      return json.dumps(data, indent=2, default=str)

• mcp_mailchimp/client.py:19-75 (helper)

  MailchimpClient class provides the post() method that create_audience uses to make the POST /lists API request.

  class MailchimpClient:
      """Lightweight async client for the Mailchimp Marketing API v3."""
  
      def __init__(self, api_key: str):
          if "-" not in api_key:
              raise ValueError(
                  "Invalid API key format. Expected: xxxxxxxxxx-usXX"
              )
          self.api_key = api_key
          self.dc = api_key.rsplit("-", 1)[-1]
          self.base_url = f"https://{self.dc}.api.mailchimp.com/3.0"
          self._client = httpx.AsyncClient(
              base_url=self.base_url,
              auth=("apikey", api_key),
              headers={"Accept": "application/json"},
              timeout=30.0,
          )
  
      @staticmethod
      def subscriber_hash(email: str) -> str:
          """MD5 hash of lowercase email — Mailchimp's subscriber identifier."""
          return hashlib.md5(email.lower().strip().encode()).hexdigest()
  
      async def _request(self, method: str, path: str, **kwargs: Any) -> Any:
          resp = await self._client.request(method, path, **kwargs)
          if resp.status_code == 204:
              return {"success": True}
          try:
              data = resp.json()
          except Exception:
              data = {"title": "Parse Error", "detail": resp.text}
          if resp.status_code >= 400:
              raise MailchimpError(
                  data.get("title", "Unknown Error"),
                  data.get("detail", "No details provided"),
                  resp.status_code,
              )
          return data
  
      async def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
          return await self._request("GET", path, params=params)
  
      async def post(self, path: str, json: dict[str, Any] | None = None) -> Any:
          return await self._request("POST", path, json=json or {})
  
      async def patch(self, path: str, json: dict[str, Any]) -> Any:
          return await self._request("PATCH", path, json=json)
  
      async def put(self, path: str, json: dict[str, Any]) -> Any:
          return await self._request("PUT", path, json=json)
  
      async def delete(self, path: str) -> Any:
          return await self._request("DELETE", path)
  
      async def close(self) -> None:
          await self._client.aclose()