Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/mailbox_tools.py:139-156 (handler)

  Primary MCP tool handler for 'create_mailbox'. Decorated with @mcp.tool and handles bulk creation by delegating to process_create_mailbox.

  @mcp.tool(
      annotations={
          "readOnlyHint": False,
          "destructiveHint": False,
          "idempotentHint": False,
          "openWorldHint": True,
      },
  )
  async def create_mailbox(
      mailboxes: List[Dict[str, Any]], ctx: Context
  ) -> Dict[str, Any]:
      """Create email mailboxes. List of dicts with: target (email/local), name (display), password (optional), recovery_email (optional), is_internal (optional), forwarding_to (optional)."""
      count = len(list(ensure_iterable(mailboxes)))
      await log_bulk_operation_start(ctx, "Creating", count, "mailbox")
  
      result = await process_create_mailbox(mailboxes, ctx)
      await log_bulk_operation_result(ctx, "Mailbox creation", result, "mailbox")
      return result

• migadu_mcp/utils/schemas.py:61-94 (schema)

  Pydantic schema MailboxCreateRequest used for input validation in create_mailbox processing via bulk_processor_with_schema decorator.

  class MailboxCreateRequest(BaseModel):
      """Request schema for creating a mailbox"""
  
      target: str = Field(..., description="Email address or local part")
      name: str = Field(..., description="Display name for the mailbox")
      password: Optional[str] = Field(
          None, description="Password (required if password_method is 'password')"
      )
      password_method: PasswordMethod = Field(
          PasswordMethod.PASSWORD, description="Password setup method"
      )
      password_recovery_email: Optional[EmailStr] = Field(
          None, description="Recovery email for invitation method"
      )
      is_internal: bool = Field(False, description="Restrict to internal-only messages")
      forwarding_to: Optional[EmailStr] = Field(
          None, description="External forwarding address"
      )
  
      @model_validator(mode="after")
      def validate_password_method(self):
          if self.password_method == PasswordMethod.INVITATION:
              if not self.password_recovery_email:
                  raise ValueError(
                      "password_recovery_email is required when password_method is invitation"
                  )
          elif self.password_method == PasswordMethod.PASSWORD:
              if not self.password:
                  raise ValueError(
                      "password is required when password_method is password"
                  )
  
          return self

• migadu_mcp/tools/mailbox_tools.py:100-138 (helper)

  Helper function process_create_mailbox that validates input with MailboxCreateRequest schema and calls the mailbox service.

  @bulk_processor_with_schema(MailboxCreateRequest)
  async def process_create_mailbox(
      validated_item: MailboxCreateRequest, ctx: Context
  ) -> Dict[str, Any]:
      """Process a single mailbox creation with Pydantic validation"""
      # Use validated Pydantic model directly - all validation already done
      target = validated_item.target
      name = validated_item.name
      password = validated_item.password
      password_recovery_email = validated_item.password_recovery_email
      is_internal = validated_item.is_internal
      forwarding_to = validated_item.forwarding_to
  
      # Parse target
      parsed = parse_email_target(target)
      domain, local_part = parsed[0]
      email_address = format_email_address(domain, local_part)
  
      await log_operation_start(ctx, "Creating mailbox", f"{email_address} ({name})")
  
      service = get_service_factory().mailbox_service()
      result = await service.create_mailbox(
          domain,
          local_part,
          name,
          password,
          password_recovery_email,
          is_internal,
          forwarding_to,
      )
  
      await log_operation_success(ctx, "Created mailbox", email_address)
      if forwarding_to:
          await ctx.info(f"🔄 Configured forwarding to: {forwarding_to}")
      if is_internal:
          await ctx.info("🔒 Configured as internal-only mailbox")
  
      return {"mailbox": result, "email_address": email_address, "success": True}

• migadu_mcp/services/mailbox_service.py:39-69 (helper)

  Core service implementation that makes the actual Migadu API POST request to create a mailbox.

  async def create_mailbox(
      self,
      domain: str,
      local_part: str,
      name: str,
      password: Optional[str] = None,
      password_recovery_email: Optional[str] = None,
      is_internal: bool = False,
      forwarding_to: Optional[str] = None,
  ) -> Dict[str, Any]:
      """Create a new email mailbox with full storage and authentication capabilities.
  
      Supports immediate password setup or invitation-based setup where the user sets their own password.
      Can create internal-only mailboxes (restricted to Migadu servers) and configure automatic
      forwarding during creation. Returns complete mailbox configuration upon successful creation.
      """
      data = {"local_part": local_part, "name": name, "is_internal": is_internal}
  
      if password:
          data["password"] = password
      elif password_recovery_email:
          data["password_method"] = "invitation"  # nosec B105 - API parameter, not a password
          data["password_recovery_email"] = password_recovery_email
  
      if forwarding_to:
          data["forwarding_to"] = forwarding_to
  
      return await self.client.request(
          "POST", f"/domains/{domain}/mailboxes", json=data
      )

• migadu_mcp/main.py:20-26 (registration)

  Registration point where register_mailbox_tools(mcp) is called, which defines and registers the create_mailbox tool via @mcp.tool decorator.

  # Register all tools
  register_mailbox_tools(mcp)
  register_identity_tools(mcp)
  register_alias_tools(mcp)
  register_rewrite_tools(mcp)
  register_resources(mcp)