Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/mailbox_tools.py:333-354 (handler)

  The main handler function for the 'set_autoresponder' tool, decorated with @mcp.tool for registration and execution. It handles bulk operations by delegating to the process helper.

  @mcp.tool(
      annotations={
          "readOnlyHint": False,
          "destructiveHint": False,
          "idempotentHint": True,
          "openWorldHint": True,
      },
  )
  async def set_autoresponder(
      autoresponders: List[Dict[str, Any]], ctx: Context
  ) -> Dict[str, Any]:
      """Configure mailbox autoresponders. List of dicts with: target (email/local), active (required), subject (optional), body (optional), expires_on (optional)."""
      count = len(list(ensure_iterable(autoresponders)))
      await log_bulk_operation_start(
          ctx, "Configuring autoresponders for", count, "mailbox"
      )
  
      result = await process_set_autoresponder(autoresponders, ctx)
      await log_bulk_operation_result(
          ctx, "Autoresponder configuration", result, "mailbox"
      )
      return result

• migadu_mcp/utils/schemas.py:132-147 (schema)

  Pydantic schema AutoresponderRequest for input validation of autoresponder configuration parameters.

  class AutoresponderRequest(BaseModel):
      """Request schema for setting autoresponder"""
  
      target: str = Field(..., description="Email address or local part")
      active: bool = Field(..., description="Whether autoresponder is enabled")
      subject: Optional[str] = Field(None, description="Subject line for replies")
      body: Optional[str] = Field(None, description="Message content for replies")
      expires_on: Optional[date] = Field(None, description="Expiration date YYYY-MM-DD")
  
      @field_validator("expires_on")
      @classmethod
      def validate_expires_on(cls, v: Optional[date]) -> Optional[date]:
          if v and v <= date.today():
              raise ValueError("expires_on must be a future date")
          return v

• migadu_mcp/tools/mailbox_tools.py:297-332 (helper)

  Helper function that processes individual autoresponder requests after schema validation, parses targets, logs operations, and calls the mailbox service.

  @bulk_processor_with_schema(AutoresponderRequest)
  async def process_set_autoresponder(
      validated_item: AutoresponderRequest, ctx: Context
  ) -> Dict[str, Any]:
      """Process a single autoresponder configuration with Pydantic validation"""
      # Use validated Pydantic model directly - all validation already done
      target = validated_item.target
      active = validated_item.active
      subject = validated_item.subject
      body = validated_item.body
      expires_on = (
          validated_item.expires_on.isoformat() if validated_item.expires_on else None
      )
  
      # Parse target
      parsed = parse_email_target(target)
      domain, local_part = parsed[0]
      email_address = format_email_address(domain, local_part)
  
      status = "Enabling" if active else "Disabling"
      await log_operation_start(ctx, f"{status} autoresponder", email_address)
  
      service = get_service_factory().mailbox_service()
      result = await service.set_autoresponder(
          domain, local_part, active, subject, body, expires_on
      )
  
      await log_operation_success(
          ctx, f"{status.lower()} autoresponder", email_address
      )
      return {
          "autoresponder": result,
          "email_address": email_address,
          "success": True,
      }

• migadu_mcp/services/mailbox_service.py:166-186 (helper)

  Service method that constructs the API payload and makes the PUT request to the Migadu API to set the autoresponder configuration.

  async def set_autoresponder(
      self,
      domain: str,
      local_part: str,
      active: bool,
      subject: Optional[str] = None,
      body: Optional[str] = None,
      expires_on: Optional[str] = None,
  ) -> Dict[str, Any]:
      """Configure mailbox autoresponder"""
      data: Dict[str, Any] = {"autorespond_active": active}
      if subject:
          data["autorespond_subject"] = subject
      if body:
          data["autorespond_body"] = body
      if expires_on:
          data["autorespond_expires_on"] = expires_on
  
      return await self.client.request(
          "PUT", f"/domains/{domain}/mailboxes/{local_part}", json=data
      )