Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/identity_tools.py:194-212 (handler)

  Main MCP tool handler for 'delete_identity', decorated with @mcp.tool(). Handles bulk deletion by calling process_delete_identity helper.

  @mcp.tool(
      annotations={
          "readOnlyHint": False,
          "destructiveHint": True,
          "idempotentHint": True,
          "openWorldHint": True,
      },
  )
  async def delete_identity(
      targets: List[Dict[str, Any]], ctx: Context
  ) -> Dict[str, Any]:
      """Delete identities. DESTRUCTIVE: Cannot be undone. List of dicts with: target, mailbox (required), domain (optional)."""
      count = len(list(ensure_iterable(targets)))
      await log_bulk_operation_start(ctx, "Deleting", count, "identity")
      await ctx.warning("🗑️ DESTRUCTIVE: This operation cannot be undone!")
  
      result = await process_delete_identity(targets, ctx)
      await log_bulk_operation_result(ctx, "Identity deletion", result, "identity")
      return result

• migadu_mcp/utils/schemas.py:224-229 (schema)

  Pydantic schema for input validation of delete_identity requests, used by bulk_processor_with_schema.

  class IdentityDeleteRequest(BaseModel):
      """Request schema for deleting an identity"""
  
      target: str = Field(..., description="Local part of identity address")
      mailbox: str = Field(..., description="Username of mailbox that owns this identity")
      domain: Optional[str] = Field(None, description="Domain name")

• migadu_mcp/tools/identity_tools.py:168-192 (helper)

  Helper function for processing individual validated delete requests, handles domain resolution, logging, and service call.

  @bulk_processor_with_schema(IdentityDeleteRequest)
  async def process_delete_identity(
      validated_item: IdentityDeleteRequest, ctx: Context
  ) -> Dict[str, Any]:
      """Process a single identity deletion"""
      # Get domain if not provided
      domain = validated_item.domain
      if domain is None:
          from migadu_mcp.config import get_config
  
          config = get_config()
          domain = config.get_default_domain()
          if not domain:
              raise ValueError("No domain provided and MIGADU_DOMAIN not configured")
  
      email_address = format_email_address(domain, validated_item.target)
      await ctx.warning(f"🗑️ DESTRUCTIVE: Deleting identity {email_address}")
  
      service = get_service_factory().identity_service()
      await service.delete_identity(
          domain, validated_item.mailbox, validated_item.target
      )
  
      await log_operation_success(ctx, "Deleted identity", email_address)
      return {"deleted": email_address, "success": True}

• migadu_mcp/services/identity_service.py:63-69 (helper)

  Service method that performs the actual HTTP DELETE request to Migadu API to delete the identity.

  async def delete_identity(
      self, domain: str, mailbox: str, identity: str
  ) -> Dict[str, Any]:
      """Delete an identity"""
      return await self.client.request(
          "DELETE", f"/domains/{domain}/mailboxes/{mailbox}/identities/{identity}"
      )